first commit

This commit is contained in:
PaulVua
2025-01-09 14:09:21 +01:00
parent 531f0ef740
commit 3081e43a1a
96 changed files with 65961 additions and 1 deletions

87
SARA/MQTT/get_config.py Executable file
View File

@@ -0,0 +1,87 @@
'''
Script to see get the SARA R4 MQTT config
ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/MQTT/get_config.py ttyAMA2 2
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0] # ex: ttyAMA2
timeout = float(parameter[1]) # ex:2
yellow_color = "\033[33m" # ANSI escape code for yellow
red_color = "\033[31m" # ANSI escape code for red
reset_color = "\033[0m" # Reset color to default
green_color = "\033[32m" # Green
#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 {}
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
while True:
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
elif time.time() > end_time:
break
time.sleep(0.1) # Short sleep to prevent busy waiting
# Decode the response and filter out empty lines
decoded_response = response.decode('utf-8')
non_empty_lines = "\n".join(line for line in decoded_response.splitlines() if line.strip())
# Add yellow color to the output
colored_output = f"{yellow_color}{non_empty_lines}\n{reset_color}"
return colored_output
# 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
)
ser.write(b'AT+UMQTT?\r') #General Information
try:
response_SARA_1 = read_complete_response(ser)
print(response_SARA_1)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

90
SARA/MQTT/login_logout.py Executable file
View File

@@ -0,0 +1,90 @@
'''
Script to see get the SARA R4 MQTT config
ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/MQTT/login_logout.py ttyAMA2 1 2
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0] # ex: ttyAMA2
login_logout = int(parameter[1]) # ex:1
timeout = float(parameter[2]) # ex:2
yellow_color = "\033[33m" # ANSI escape code for yellow
red_color = "\033[31m" # ANSI escape code for red
reset_color = "\033[0m" # Reset color to default
green_color = "\033[32m" # Green
#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 {}
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
while True:
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
elif time.time() > end_time:
break
time.sleep(0.1) # Short sleep to prevent busy waiting
# Decode the response and filter out empty lines
decoded_response = response.decode('utf-8')
non_empty_lines = "\n".join(line for line in decoded_response.splitlines() if line.strip())
# Add yellow color to the output
colored_output = f"{yellow_color}{non_empty_lines}\n{reset_color}"
return colored_output
# 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
)
command = f'AT+UMQTTC={login_logout}\r'
ser.write((command + '\r').encode('utf-8'))
try:
response_SARA_1 = read_complete_response(ser)
print(response_SARA_1)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

89
SARA/MQTT/publish.py Executable file
View File

@@ -0,0 +1,89 @@
'''
Script to publish a message via MQTT
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/MQTT/publish.py ttyAMA2 Hello 2
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
message = parameter[1] # ex: Hello
timeout = float(parameter[2]) # ex:2
yellow_color = "\033[33m" # ANSI escape code for yellow
red_color = "\033[31m" # ANSI escape code for red
reset_color = "\033[0m" # Reset color to default
green_color = "\033[32m" # Green
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
while True:
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
elif time.time() > end_time:
break
time.sleep(0.1) # Short sleep to prevent busy waiting
# Decode the response and filter out empty lines
decoded_response = response.decode('utf-8')
non_empty_lines = "\n".join(line for line in decoded_response.splitlines() if line.strip())
# Add yellow color to the output
colored_output = f"{yellow_color}{non_empty_lines}\n{reset_color}"
return colored_output
#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
)
command = f'AT+UMQTTC=2,0,0,"notif","{message}"\r'
ser.write((command + '\r').encode('utf-8'))
try:
response_SARA_1 = read_complete_response(ser)
print(response_SARA_1)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

94
SARA/MQTT/set_config.py Executable file
View File

@@ -0,0 +1,94 @@
'''
Script to connect set the MQTT config
1: local TCP port number
2: server name
3: server IP addr
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/MQTT/set_config.py ttyAMA2 2 aircarto.fr 2
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
config_id = parameter[1] # ex: 1
config_value = parameter[2] # ex: aircarto.fr
timeout = float(parameter[3]) # ex:2
yellow_color = "\033[33m" # ANSI escape code for yellow
red_color = "\033[31m" # ANSI escape code for red
reset_color = "\033[0m" # Reset color to default
green_color = "\033[32m" # Green
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
while True:
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
elif time.time() > end_time:
break
time.sleep(0.1) # Short sleep to prevent busy waiting
# Decode the response and filter out empty lines
decoded_response = response.decode('utf-8')
non_empty_lines = "\n".join(line for line in decoded_response.splitlines() if line.strip())
# Add yellow color to the output
colored_output = f"{yellow_color}{non_empty_lines}\n{reset_color}"
return colored_output
#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
)
command = f'AT+UMQTT={config_id},"{config_value}"\r'
ser.write((command + '\r').encode('utf-8'))
try:
response_SARA_1 = read_complete_response(ser)
print(response_SARA_1)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

82
SARA/sara.py Executable file
View File

@@ -0,0 +1,82 @@
'''
Script to see if the SARA-R410 is running
ex:
python3 /var/www/nebuleair_pro_4g/SARA/sara.py ttyAMA2 AT+CCID? 2
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0] # ex: ttyAMA2
command = parameter[1] # ex: AT+CCID?
timeout = float(parameter[2]) # 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
)
ser.write((command + '\r').encode('utf-8'))
#ser.write(b'ATI\r') #General Information
#ser.write(b'AT+CCID?\r') #SIM card number
#ser.write(b'AT+CPIN?\r') #Check the status of the SIM card
#ser.write(b'AT+CIND?\r') #Indication state (last number is SIM detection: 0 no SIM detection, 1 SIM detected, 2 not available)
#ser.write(b'AT+UGPIOR=?\r') #Reads the current value of the specified GPIO pin
#ser.write(b'AT+UGPIOC?\r') #GPIO select configuration
#ser.write(b'AT+COPS=?\r') #Check the network and cellular technology the modem is currently using
#ser.write(b'AT+COPS=1,2,20801') #connext to orange
#ser.write(b'AT+CFUN=?\r') #Selects/read the level of functionality
#ser.write(b'AT+URAT=?\r') #Radio Access Technology
#ser.write(b'AT+USIMSTAT?')
#ser.write(b'AT+IPR=115200') #Check/Define baud rate
#ser.write(b'AT+CMUX=?')
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

70
SARA/sara_connectNetwork.py Executable file
View File

@@ -0,0 +1,70 @@
'''
Script to connect SARA-R410 to network SARA-R410
AT+COPS=1,2,20801
mode->1 pour manual
format->2 pour numeric
operator->20801 pour orange
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
networkID = parameter[1] # ex: 20801
timeout = float(parameter[2]) # 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
)
command = f'AT+COPS=1,2,"{networkID}"\r'
ser.write((command + '\r').encode('utf-8'))
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

65
SARA/sara_eraseMessage.py Executable file
View File

@@ -0,0 +1,65 @@
'''
Script to erase memory to SARA-R410 memory
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
message = parameter[1] # ex: Hello
#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 = 2
)
#0. empty file
ser.write(b'AT+UDELFILE="sensordata.json"\r')
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

62
SARA/sara_readMessage.py Executable file
View File

@@ -0,0 +1,62 @@
'''
Script to read memory to SARA-R410 memory
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
message = parameter[1] # ex: Hello
#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 = 2
)
ser.write(b'AT+URDFILE="sensordata.json"\r')
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

63
SARA/sara_sendMessage.py Executable file
View File

@@ -0,0 +1,63 @@
'''
Script to send message to URL with SARA-R410
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
endpoint = parameter[1] # ex: /pro_4G/notif_message.php
#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 = 2
)
command= f'AT+UHTTPC=0,4,"{endpoint}","data.txt","sensordata.json",4\r'
ser.write((command + '\r').encode('utf-8'))
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

69
SARA/sara_setURL.py Executable file
View File

@@ -0,0 +1,69 @@
'''
Script to set the URL for a HTTP request
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0] # ex: ttyAMA2
url = parameter[1] # ex: data.mobileair.fr
#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 = 2
)
command = f'AT+UHTTP=0,1,"{url}"\r'
ser.write((command + '\r').encode('utf-8'))
print("****")
print("SET URL (SARA)")
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
print("****")
#print("Serial closed")

71
SARA/sara_writeMessage.py Executable file
View File

@@ -0,0 +1,71 @@
'''
Script to write message to SARA-R410 memory
'''
import serial
import time
import sys
import json
parameter = sys.argv[1:] # Exclude the script name
port='/dev/'+parameter[0] # ex: ttyAMA2
message = parameter[1] # ex: Hello
#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 = 2
)
#0. empty file
#ser.write(b'AT+UDELFILE="sensordata.json"\r')
#1. Open sensordata.json (with correct data size)
size_of_string = len(message)
command = f'AT+UDWNFILE="sensordata.json",{size_of_string}\r'
ser.write((command + '\r').encode('utf-8'))
time.sleep(1)
#2. Write to shell
ser.write(message.encode())
try:
# Read lines until a timeout occurs
response_lines = []
while True:
line = ser.readline().decode('utf-8').strip()
if not line:
break # Break the loop if an empty line is encountered
response_lines.append(line)
# Print the response
for line in response_lines:
print(line)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")