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:
Your Name
2025-01-28 18:14:55 +01:00
parent 083d342373
commit d095e53cd6
10 changed files with 646 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
NPM/firmware_version.py Normal file
View File

@@ -0,0 +1,64 @@
'''
Script to get NPM firmware version
need parameter: port
/usr/bin/python3 /var/www/nebuleair_pro_4g/NPM/firmware_version.py ttyAMA5
'''
import serial
import requests
import json
import sys
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0]
ser = serial.Serial(
port=port,
baudrate=115200,
parity=serial.PARITY_EVEN,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
timeout = 1
)
ser.write(b'\x81\x17\x68') #firmware version
while True:
try:
byte_data = ser.readline()
formatted = ''.join(f'\\x{byte:02x}' for byte in byte_data)
#print(formatted)
'''
la réponse est de type
\x81\x17\x00\x10\x46\x12
avec
\x81 address
\x17 command code
\x00 state
\x10\x46 firmware version
\x12 checksum
'''
# Extract byte 4 and byte 5
byte4 = byte_data[3] # 4th byte (index 3)
byte5 = byte_data[4] # 5th byte (index 4)
firmware_version = int(f"{byte4:02x}{byte5:02x}")
data = {
'firmware_version': firmware_version,
}
json_data = json.dumps(data)
print(json_data)
break
except KeyboardInterrupt:
print("User interrupt encountered. Exiting...")
time.sleep(3)
exit()
except:
# for all other kinds of error, but not specifying which one
print("Unknown error...")
time.sleep(3)
exit()

4
npm.py → NPM/get_data.py
View File

@@ -1,7 +1,7 @@
'''
Script to get NPM values
need parameter: port
/usr/bin/python3 /var/www/nebuleair_pro_4g/npm.py ttyAMA4
/usr/bin/python3 /var/www/nebuleair_pro_4g/NPM/get_data.py ttyAMA5
'''
import serial
@@ -19,7 +19,7 @@ ser = serial.Serial(
parity=serial.PARITY_EVEN,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
timeout = 2
timeout = 1
)
ser.write(b'\x81\x11\x6E') #data10s

103
NPM/get_data_modbus.py Normal file
View File

@@ -0,0 +1,103 @@
'''
Script to get NPM data via Modbus
need parameter: port
/usr/bin/python3 /var/www/nebuleair_pro_4g/NPM/get_data_modbus.py ttyAMA5
Modbus RTU
[Slave Address][Function Code][Starting Address][Quantity of Registers][CRC]
Pour récupérer les 5 cannaux (a partir du registre 0x80)
Request
\x01\x03\x00\x80\x00\x0A\xE4\x1E
\x01 Slave Address (slave device address)
\x03 Function code (read multiple holding registers)
\x00\x80 Starting Address (The request starts reading from holding register address 0x80 or 128)
\x00\x0A Quantity of Registers (Requests to read 0x0A or 10 consecutive registers starting from address 50)
\xE4\x1E Cyclic Redundancy Check (checksum )
'''
import serial
import requests
import json
import sys
import crcmod
parameter = sys.argv[1:] # Exclude the script name
#print("Parameters received:")
port='/dev/'+parameter[0]
ser = serial.Serial(
port=port,
baudrate=115200,
parity=serial.PARITY_EVEN,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
timeout = 0.5
)
# Define Modbus CRC-16 function
crc16 = crcmod.predefined.mkPredefinedCrcFun('modbus')
# Request frame without CRC
data = b'\x01\x03\x00\x80\x00\x0A'
# Calculate CRC
crc = crc16(data)
crc_low = crc & 0xFF
crc_high = (crc >> 8) & 0xFF
# Append CRC to the frame
request = data + bytes([crc_low, crc_high])
print(f"Request frame: {request.hex()}")
ser.write(request)
while True:
try:
byte_data = ser.readline()
formatted = ''.join(f'\\x{byte:02x}' for byte in byte_data)
print(formatted)
# Extract LSW (first 2 bytes) and MSW (last 2 bytes)
lsw_channel1 = int.from_bytes(byte_data[3:5], byteorder='little')
msw_chanel1 = int.from_bytes(byte_data[5:7], byteorder='little')
raw_value_channel1 = (msw_chanel1 << 16) | lsw_channel1
lsw_channel2 = int.from_bytes(byte_data[7:9], byteorder='little')
msw_chanel2 = int.from_bytes(byte_data[9:11], byteorder='little')
raw_value_channel2 = (msw_chanel2 << 16) | lsw_channel2
lsw_channel3 = int.from_bytes(byte_data[11:13], byteorder='little')
msw_chanel3 = int.from_bytes(byte_data[13:15], byteorder='little')
raw_value_channel3 = (msw_chanel3 << 16) | lsw_channel3
lsw_channel4 = int.from_bytes(byte_data[15:17], byteorder='little')
msw_chanel4 = int.from_bytes(byte_data[17:19], byteorder='little')
raw_value_channel4 = (msw_chanel1 << 16) | lsw_channel4
lsw_channel5 = int.from_bytes(byte_data[19:21], byteorder='little')
msw_chanel5 = int.from_bytes(byte_data[21:23], byteorder='little')
raw_value_channel5 = (msw_chanel5 << 16) | lsw_channel5
print(f"Channel 1 (0.2->0.5): {raw_value_channel1}")
print(f"Channel 2 (0.5->1.0): {raw_value_channel2}")
print(f"Channel 3 (1.0->2.5): {raw_value_channel3}")
print(f"Channel 4 (2.5->5.0): {raw_value_channel4}")
print(f"Channel 5 (5.0->10.): {raw_value_channel5}")
break
except KeyboardInterrupt:
print("User interrupt encountered. Exiting...")
time.sleep(3)
exit()
except:
# for all other kinds of error, but not specifying which one
print("Unknown error...")
time.sleep(3)
exit()

2
README.md
View File

@@ -16,7 +16,7 @@ See `installation.sh`
```
sudo apt update
sudo apt install git gh apache2 php python3 python3-pip jq autossh i2c-tools python3-smbus -y
sudo pip3 install pyserial requests RPi.GPIO adafruit-circuitpython-bme280 --break-system-packages
sudo pip3 install pyserial requests RPi.GPIO adafruit-circuitpython-bme280 crcmod --break-system-packages
sudo mkdir -p /var/www/.ssh
sudo ssh-keygen -t rsa -b 4096 -f /var/www/.ssh/id_rsa -N ""
sudo ssh-copy-id -i /var/www/.ssh/id_rsa.pub -p 50221 airlab_server1@aircarto.fr

24
SARA/SSL/full_test_HTTPS_POST.py
View File

@@ -100,7 +100,7 @@ ser_sara = serial.Serial(
try:
#step 1: import the certificate
print("****")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/isrg-root-x2.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)
@@ -108,7 +108,7 @@ try:
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,"isrg-root-x2",{size_of_string}\r'
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(response_SARA_1)
@@ -133,7 +133,7 @@ try:
# *******************************
# SECURITY PROFILE
# AT+USECPRF=<profile_id>[,<op_code>[,<param_val>]]
security_profile_id = 0
security_profile_id = 1
# op_code: 0 -> certificate validation level
@@ -147,9 +147,9 @@ try:
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;
# param_val : 0 -> any; server can use any version for the connection; 1-> LSv1.0; 2->TLSv1.1; 3->TLSv1.2;
print("\033[0;33mSet the security profile (params)\033[0m")
minimum_SSL_version = 0
minimum_SSL_version = 3
command = f'AT+USECPRF={security_profile_id},1,{minimum_SSL_version}\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5bb = read_complete_response(ser_sara, wait_for_line="OK")
@@ -157,8 +157,9 @@ try:
time.sleep(0.5)
#op_code: 2 -> cipher suite
# 0 (factory-programmed value): (0x0000) Automatic the cipher suite will be negotiated in the handshake process
print("\033[0;33mSet cipher \033[0m")
cipher_suite = 1
cipher_suite = 0
command = f'AT+USECPRF={security_profile_id},2,{cipher_suite}\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5cc = read_complete_response(ser_sara, wait_for_line="OK")
@@ -167,12 +168,21 @@ try:
# 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,"isrg-root-x2"\r'
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)
# op_code: 10 -> SNI (server name indication)
print("\033[0;33mSet the SNI\033[0m")
command = f'AT+USECPRF={security_profile_id},10,"{url}"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5cf = read_complete_response(ser_sara, wait_for_line="OK")
print(response_SARA_5cf)
time.sleep(0.5)
# *************************
# *************************

325
SARA/SSL/test_22.py Normal file
View File

@@ -0,0 +1,325 @@
'''
Script to set the URL for a HTTP request and trigger the POST Request
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 api-prod.uspot.probesys.net /nebuleair?token=2AFF6dQk68daFZ
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 webhook.site /6bee2237-099a-4ff4-8452-9f4126df7151
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 aircarto.fr /tests/test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 ssl.aircarto.fr /test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 vps.aircarto.fr /test.php
First profile id:
AT+UHTTP=0,1,"data.nebuleair.fr"
Second profile id:
AT+UHTTP=1,1,"api-prod.uspot.probesys.net"
Third profile id:
AT+UHTTP=2,1,"aircarto.fr"
'''
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
endpoint = parameter[2]
profile_id = 3
#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)
send_uSpot = config.get('send_uSpot', False)
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(
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
)
try:
#step 1: import the certificate
print("****")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/isrgrootx1.der", "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,"isrgrootx1",{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)
#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>]]
security_profile_id = 2
# op_code: 0 -> certificate validation level
# param_val : 0 -> Level 0 No validation; 1-> Level 1 Root certificate validation
print("\033[0;33mSet the security profile (params)\033[0m")
certification_level=1
command = f'AT+USECPRF={security_profile_id},0,{certification_level}\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: 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,"isrgrootx1"\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("\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, wait_for_line="OK")
print(response_SARA_5)
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={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_line="OK")
print(response_SARA_5)
time.sleep(1)
# Write Data to saraR4
payload_json = {
"nebuleairid": "C04F8B8D3A08",
"software_version": "ModuleAir-V1-012023",
"sensordatavalues": [
{
"value_type": "NPM_P0",
"value": "2.3"
},
{
"value_type": "NPM_P0",
"value": "3.30"
},
{
"value_type": "NPM_P1",
"value": "9.05"
},
{
"value_type": "NPM_P2",
"value": "20.60"
},
{
"value_type": "NPM_N1",
"value": "49.00"
},
{
"value_type": "NPM_N10",
"value": "49.00"
},
{
"value_type": "NPM_N25",
"value": "49.00"
},
{
"value_type": "BME280_temperature",
"value": "25.82"
}
]
}
# 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.encode('utf-8'))
response_SARA_1 = read_complete_response(ser_sara, wait_for_line=">")
print(response_SARA_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, 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("\033[0;33mPOST REQUEST\033[0m")
#parameter (POST)
command = f'AT+UHTTPC={profile_id},4,"{endpoint}","https.resp","sensordata_json.json",4\r'
#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_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("\033[0;33mREPLY SERVER\033[0m")
ser_sara.write(b'AT+URDFILE="https.resp"\r')
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("\033[0;33mEmpty Memory\033[0m")
ser_sara.write(b'AT+UDELFILE="sensordata_json.json"\r')
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 (for error_class 3)
0 No error
11 Server connection error
73 Secure socket connect error
'''
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

133
SARA/SSL/test_33.py Normal file
View File

@@ -0,0 +1,133 @@
'''
Script to set the URL for a HTTP request and trigger the POST Request
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 api-prod.uspot.probesys.net /nebuleair?token=2AFF6dQk68daFZ
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 webhook.site /6bee2237-099a-4ff4-8452-9f4126df7151
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 aircarto.fr /tests/test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 ssl.aircarto.fr /test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 vps.aircarto.fr /test.php
First profile id:
AT+UHTTP=0,1,"data.nebuleair.fr"
Second profile id:
AT+UHTTP=1,1,"api-prod.uspot.probesys.net"
Third profile id:
AT+UHTTP=2,1,"aircarto.fr"
'''
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
endpoint = parameter[2]
profile_id = 3
#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)
send_uSpot = config.get('send_uSpot', False)
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(
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
)
try:
#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 layer
#1
print("\033[0;33mCheck certificate\033[0m")
command = f'AT+USECPRF=<profile_id>,<op_code>\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)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

2
html/launcher.php
View File

@@ -63,7 +63,7 @@ if ($type == "reboot") {
if ($type == "npm") {
$port=$_GET['port'];
$command = '/usr/bin/python3 /var/www/nebuleair_pro_4g/npm.py ' . $port;
$command = '/usr/bin/python3 /var/www/nebuleair_pro_4g/NPM/get_data.py ' . $port;
$output = shell_exec($command);
echo $output;
}

0
index_old.html → old/index_old.html
View File

0
launcher.php → old/launcher.php
View File