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

update

Browse Source
This commit is contained in:
root
2025-02-20 15:07:49 +01:00
commit 59133628af
228 changed files with 282657 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
SARA/SSL/certificate/e5.der Normal file
View File

Binary file not shown.

17
SARA/SSL/certificate/e5.pem Normal file
View File

@@ -0,0 +1,17 @@
-----BEGIN CERTIFICATE-----
MIICtDCCAjugAwIBAgIQGG511O6woF39Lagghl0eMTAKBggqhkjOPQQDAzBPMQsw
CQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJuZXQgU2VjdXJpdHkgUmVzZWFyY2gg
R3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBYMjAeFw0yNDAzMTMwMDAwMDBaFw0y
NzAzMTIyMzU5NTlaMDIxCzAJBgNVBAYTAlVTMRYwFAYDVQQKEw1MZXQncyBFbmNy
eXB0MQswCQYDVQQDEwJFNTB2MBAGByqGSM49AgEGBSuBBAAiA2IABA0LOoprYY62
79xfWOfGQkVUq2P2ZmFICi5ZdbSBAjdQtz8WedyY7KEol3IgHCzP1XxSIE5UeFuE
FGvAkK6F7MBRQTxah38GTdT+YNH6bC3hfZUQiKIIVA+ZGkzm6gqs2KOB+DCB9TAO
BgNVHQ8BAf8EBAMCAYYwHQYDVR0lBBYwFAYIKwYBBQUHAwIGCCsGAQUFBwMBMBIG
A1UdEwEB/wQIMAYBAf8CAQAwHQYDVR0OBBYEFJ8rX888IU+dBLftKyzExnCL0tcN
MB8GA1UdIwQYMBaAFHxClq7eS0g7+pL4nozPbYupcjeVMDIGCCsGAQUFBwEBBCYw
JDAiBggrBgEFBQcwAoYWaHR0cDovL3gyLmkubGVuY3Iub3JnLzATBgNVHSAEDDAK
MAgGBmeBDAECATAnBgNVHR8EIDAeMBygGqAYhhZodHRwOi8veDIuYy5sZW5jci5v
cmcvMAoGCCqGSM49BAMDA2cAMGQCMBttLkVBHEU+2V80GHRnE3m6qym1thBOgydK
i0VOx3vP9EAwHWGl5hxtpJAJkm5GSwIwRikYhDR6vPve2BvYGacE9ct+522E2dqO
6s42MLmigEws5mASS6l2quhtlUfacgkM
-----END CERTIFICATE-----

BIN
SARA/SSL/certificate/e6.der Normal file
View File

Binary file not shown.

17
SARA/SSL/certificate/e6.pem Normal file
View File

@@ -0,0 +1,17 @@
-----BEGIN CERTIFICATE-----
MIICtjCCAjygAwIBAgIRAICpc0jvJ2ip4/a7Q8D5xikwCgYIKoZIzj0EAwMwTzEL
MAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2VhcmNo
IEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDIwHhcNMjQwMzEzMDAwMDAwWhcN
MjcwMzEyMjM1OTU5WjAyMQswCQYDVQQGEwJVUzEWMBQGA1UEChMNTGV0J3MgRW5j
cnlwdDELMAkGA1UEAxMCRTYwdjAQBgcqhkjOPQIBBgUrgQQAIgNiAATZ8Z5Gh/gh
cWCoJuuj+rnq2h25EqfUJtlRFLFhfHWWvyILOR/VvtEKRqotPEoJhC6+QJVV6RlA
N2Z17TJOdwRJ+HB7wxjnzvdxEP6sdNgA1O1tHHMWMxCcOrLqbGL0vbijgfgwgfUw
DgYDVR0PAQH/BAQDAgGGMB0GA1UdJQQWMBQGCCsGAQUFBwMCBggrBgEFBQcDATAS
BgNVHRMBAf8ECDAGAQH/AgEAMB0GA1UdDgQWBBSTJ0aYA6lRaI6Y1sRCSNsjv1iU
0jAfBgNVHSMEGDAWgBR8Qpau3ktIO/qS+J6Mz22LqXI3lTAyBggrBgEFBQcBAQQm
MCQwIgYIKwYBBQUHMAKGFmh0dHA6Ly94Mi5pLmxlbmNyLm9yZy8wEwYDVR0gBAww
CjAIBgZngQwBAgEwJwYDVR0fBCAwHjAcoBqgGIYWaHR0cDovL3gyLmMubGVuY3Iu
b3JnLzAKBggqhkjOPQQDAwNoADBlAjBgGMvAszhCd1BsRuMwGYCC0QCzf5d//MC5
ASqIyswj3hGcoZREOKDKdvJPHhgdZr8CMQCWq4Kjl/RmuF49LBq9eP7oGWAc55w4
G72FoKw5a9WywSwBzoJunrayTB8nDSjEhu8=
-----END CERTIFICATE-----

BIN
SARA/SSL/certificate/isrg-root-x2.der Normal file
View File

Binary file not shown.

BIN
SARA/SSL/certificate/isrgrootx1.der Normal file
View File

Binary file not shown.

31
SARA/SSL/certificate/isrgrootx1.pem Normal file
View File

@@ -0,0 +1,31 @@
-----BEGIN CERTIFICATE-----
MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw
TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4
WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu
ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY
MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc
h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+
0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U
A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW
T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH
B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC
B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv
KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn
OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn
jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw
qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI
rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV
HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq
hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL
ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ
3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK
NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5
ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur
TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC
jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc
oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq
4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA
mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d
emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=
-----END CERTIFICATE-----

BIN
SARA/SSL/certificate/r11.der Normal file
View File

Binary file not shown.

29
SARA/SSL/certificate/r11.pem Normal file
View File

@@ -0,0 +1,29 @@
-----BEGIN CERTIFICATE-----
MIIFBjCCAu6gAwIBAgIRAIp9PhPWLzDvI4a9KQdrNPgwDQYJKoZIhvcNAQELBQAw
TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMjQwMzEzMDAwMDAw
WhcNMjcwMzEyMjM1OTU5WjAzMQswCQYDVQQGEwJVUzEWMBQGA1UEChMNTGV0J3Mg
RW5jcnlwdDEMMAoGA1UEAxMDUjExMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB
CgKCAQEAuoe8XBsAOcvKCs3UZxD5ATylTqVhyybKUvsVAbe5KPUoHu0nsyQYOWcJ
DAjs4DqwO3cOvfPlOVRBDE6uQdaZdN5R2+97/1i9qLcT9t4x1fJyyXJqC4N0lZxG
AGQUmfOx2SLZzaiSqhwmej/+71gFewiVgdtxD4774zEJuwm+UE1fj5F2PVqdnoPy
6cRms+EGZkNIGIBloDcYmpuEMpexsr3E+BUAnSeI++JjF5ZsmydnS8TbKF5pwnnw
SVzgJFDhxLyhBax7QG0AtMJBP6dYuC/FXJuluwme8f7rsIU5/agK70XEeOtlKsLP
Xzze41xNG/cLJyuqC0J3U095ah2H2QIDAQABo4H4MIH1MA4GA1UdDwEB/wQEAwIB
hjAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwEgYDVR0TAQH/BAgwBgEB
/wIBADAdBgNVHQ4EFgQUxc9GpOr0w8B6bJXELbBeki8m47kwHwYDVR0jBBgwFoAU
ebRZ5nu25eQBc4AIiMgaWPbpm24wMgYIKwYBBQUHAQEEJjAkMCIGCCsGAQUFBzAC
hhZodHRwOi8veDEuaS5sZW5jci5vcmcvMBMGA1UdIAQMMAowCAYGZ4EMAQIBMCcG
A1UdHwQgMB4wHKAaoBiGFmh0dHA6Ly94MS5jLmxlbmNyLm9yZy8wDQYJKoZIhvcN
AQELBQADggIBAE7iiV0KAxyQOND1H/lxXPjDj7I3iHpvsCUf7b632IYGjukJhM1y
v4Hz/MrPU0jtvfZpQtSlET41yBOykh0FX+ou1Nj4ScOt9ZmWnO8m2OG0JAtIIE38
01S0qcYhyOE2G/93ZCkXufBL713qzXnQv5C/viOykNpKqUgxdKlEC+Hi9i2DcaR1
e9KUwQUZRhy5j/PEdEglKg3l9dtD4tuTm7kZtB8v32oOjzHTYw+7KdzdZiw/sBtn
UfhBPORNuay4pJxmY/WrhSMdzFO2q3Gu3MUBcdo27goYKjL9CTF8j/Zz55yctUoV
aneCWs/ajUX+HypkBTA+c8LGDLnWO2NKq0YD/pnARkAnYGPfUDoHR9gVSp/qRx+Z
WghiDLZsMwhN1zjtSC0uBWiugF3vTNzYIEFfaPG7Ws3jDrAMMYebQ95JQ+HIBD/R
PBuHRTBpqKlyDnkSHDHYPiNX3adPoPAcgdF3H2/W0rmoswMWgTlLn1Wu0mrks7/q
pdWfS6PJ1jty80r2VKsM/Dj3YIDfbjXKdaFU5C+8bhfJGqU3taKauuz0wHVGT3eo
6FlWkWYtbt4pgdamlwVeZEW+LM7qZEJEsMNPrfC03APKmZsJgpWCDWOKZvkZcvjV
uYkQ4omYCTX5ohy+knMjdOmdH9c7SpqEWBDC86fiNex+O0XOMEZSa8DA
-----END CERTIFICATE-----

35
SARA/SSL/curl_script.sh Normal file
View File

@@ -0,0 +1,35 @@
#!/bin/bash
# to run this script ./curl_script.sh
# URL to send the request
URL_microSpot="http://api-prod.uspot.probesys.net:81/nebuleair?token=2AFF6dQk68daFZ"
URL_microSpot_HTTPS="https://api-prod.uspot.probesys.net:443/nebuleair?token=2AFF6dQk68daFZ"
URL_airCarto_HTTPS="https://aircarto.fr/tests/test.php"
URL_webhook="https://webhook.site/6bee2237-099a-4ff4-8452-9f4126df7151"
CERT_PATH="/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.pem"
CIPHER="TLS_AES_256_GCM_SHA384"
#CIPHER="TLS_AES_256_GCM_SHA384"
#CIPHER="POLY1305_SHA256"
# JSON payload to send
PAYLOAD='{
"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"}
]
}'
# Perform the curl command
curl --http1.1 -v -X POST "$URL_microSpot_HTTPS" \
--tlsv1.2 \
--cacert "$CERT_PATH" \
-d "$PAYLOAD"

134
SARA/SSL/full_test_HTTP.py Normal file
View File

@@ -0,0 +1,134 @@
'''
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/full_test_HTTP.py ttyAMA2 data.nebuleair.fr
To do: need to add profile id as parameter
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
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 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 4: set url (op_code = 1)
print("****")
print("SET URL")
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)
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,80\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
print(response_SARA_55)
time.sleep(1)
#step 4: trigger the request
print("****")
print("Trigger POST REQUEST")
command = f'AT+UHTTPC={profile_id},1,"/pro_4G/test.php","http.resp"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_6 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=50, wait_for_line="+UUHTTPCR")
print(response_SARA_6)
time.sleep(1)
#READ REPLY
print("****")
print("Read reply from server")
ser_sara.write(b'AT+URDFILE="http.resp"\r')
response_SARA_7 = read_complete_response(ser_sara)
print("Reply from server:")
print(response_SARA_7)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

160
SARA/SSL/full_test_HTTPS.py Normal file
View File

@@ -0,0 +1,160 @@
'''
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/full_test_HTTPS.py ttyAMA2 api-prod.uspot.probesys.net
To do: need to add profile id as parameter
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
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 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
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("****")
print("Add certificate")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.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'
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)
ser_sara.write(certificate)
response_SARA_2 = read_complete_response(ser_sara)
print("Write certificate data")
print(response_SARA_2)
time.sleep(1)
#step 4: set url (op_code = 1)
print("****")
print("SET URL")
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)
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'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
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'
ser_sara.write(command.encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
#step 4: trigger the request (http_command=1 for GET)
print("****")
print("Trigger POST REQUEST")
command = f'AT+UHTTPC={profile_id},1,"/tests/test.php","https.resp"\r'
#command = f'AT+UHTTPC={profile_id},1,"/nebuleair?token=2AFF6dQk68daFZ","https.resp"\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
#READ REPLY
print("****")
print("Read reply from server")
ser_sara.write(b'AT+URDFILE="https.resp"\r')
response_SARA_7 = read_complete_response(ser_sara)
print("Reply from server:")
print(response_SARA_7)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

358
SARA/SSL/full_test_HTTPS_POST.py Normal file
View File

@@ -0,0 +1,358 @@
'''
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/full_test_HTTPS_POST.py ttyAMA2 api-prod.uspot.probesys.net /nebuleair?token=2AFF6dQk68daFZ
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 webhook.site /0904d7b1-2558-43b9-8b35-df5bc40df967
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 aircarto.fr /tests/test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 ssl.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 = 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)
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("****")
certificate_name = "e6"
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.pem", "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,"{certificate_name}",{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 = 1
# 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=0
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: 1 -> minimum SSL/TLS version
# 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
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")
print(response_SARA_5bb)
time.sleep(0.5)
#op_code: 2 -> legacy cipher suite selection
# 0 (factory-programmed value): a list of default cipher suites is proposed at the beginning of handshake process, and a cipher suite will be negotiated among the cipher suites proposed in the list.
print("\033[0;33mSet cipher \033[0m")
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")
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,"{certificate_name}"\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)
# *************************
# *************************
#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 PORT (op_code = 5)
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, 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("\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=50, 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;33mGet error code\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 or 10)
0 No error
11 Server connection error
22 PSD or CSD connection not established
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")

207
SARA/SSL/full_test_HTTP_POST.py Normal file
View File

@@ -0,0 +1,207 @@
'''
Script to set the URL for a HTTP request and trigger the POST Request
FONCTIONNE SUR data.nebuleair.fr
FONCTIONNE SUR uSpot
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTP_POST.py ttyAMA2 api-prod.uspot.probesys.net
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTP_POST.py ttyAMA2 aircarto.fr /tests/test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTP_POST.py ttyAMA2 ssl.aircarto.fr /test.php
To do: need to add profile id as parameter
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 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
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 4: set url (op_code = 1)
print("****")
print("SET URL")
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)
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,80\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
print(response_SARA_55)
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"
}
]
}
payload_csv = [None] * 20
payload_csv[0] = 1
payload_csv[1] = 1
payload_csv[2] = 1
#csv_string = ','.join(str(value) if value is not None else '' for value in payload_csv)
#size_of_string = len(csv_string)
# 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)
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:")
print(response_SARA_1)
time.sleep(1)
#2. Write to shell
ser_sara.write(payload_string.encode())
response_SARA_2 = read_complete_response(ser_sara)
print("Write to memory:")
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},4,"{endpoint}","http.resp","sensordata_json.json",4\r'
#AirCarto
#command = f'AT+UHTTPC={profile_id},1,"/tests/test.php","http.resp"\r'
#command = f'AT+UHTTPC={profile_id},4,"/wifi.php","http.resp","sensordata_json.json",4\r'
#command = f'AT+UHTTPC={profile_id},4,"/pro_4G/data.php?sensor_id=52E7573A","http.resp","sensordata_json.json",4\r'
#AtmoSud
#command = f'AT+UHTTPC={profile_id},4,"/nebuleair?token=2AFF6dQk68daFZ","http.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
#READ REPLY
print("****")
print("Read reply from server")
ser_sara.write(b'AT+URDFILE="http.resp"\r')
response_SARA_7 = read_complete_response(ser_sara)
print("Reply from server:")
print(response_SARA_7)
#5. empty json
print("Empty SARA memory:")
ser_sara.write(b'AT+UDELFILE="sensordata_json.json"\r')
response_SARA_8 = read_complete_response(ser_sara)
print(response_SARA_8)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

35
SARA/SSL/old/ISRGRootX1.txt Normal file
View File

@@ -0,0 +1,35 @@
00:ad:e8:24:73:f4:14:37:f3:9b:9e:2b:57:28:1c:
87:be:dc:b7:df:38:90:8c:6e:3c:e6:57:a0:78:f7:
75:c2:a2:fe:f5:6a:6e:f6:00:4f:28:db:de:68:86:
6c:44:93:b6:b1:63:fd:14:12:6b:bf:1f:d2:ea:31:
9b:21:7e:d1:33:3c:ba:48:f5:dd:79:df:b3:b8:ff:
12:f1:21:9a:4b:c1:8a:86:71:69:4a:66:66:6c:8f:
7e:3c:70:bf:ad:29:22:06:f3:e4:c0:e6:80:ae:e2:
4b:8f:b7:99:7e:94:03:9f:d3:47:97:7c:99:48:23:
53:e8:38:ae:4f:0a:6f:83:2e:d1:49:57:8c:80:74:
b6:da:2f:d0:38:8d:7b:03:70:21:1b:75:f2:30:3c:
fa:8f:ae:dd:da:63:ab:eb:16:4f:c2:8e:11:4b:7e:
cf:0b:e8:ff:b5:77:2e:f4:b2:7b:4a:e0:4c:12:25:
0c:70:8d:03:29:a0:e1:53:24:ec:13:d9:ee:19:bf:
10:b3:4a:8c:3f:89:a3:61:51:de:ac:87:07:94:f4:
63:71:ec:2e:e2:6f:5b:98:81:e1:89:5c:34:79:6c:
76:ef:3b:90:62:79:e6:db:a4:9a:2f:26:c5:d0:10:
e1:0e:de:d9:10:8e:16:fb:b7:f7:a8:f7:c7:e5:02:
07:98:8f:36:08:95:e7:e2:37:96:0d:36:75:9e:fb:
0e:72:b1:1d:9b:bc:03:f9:49:05:d8:81:dd:05:b4:
2a:d6:41:e9:ac:01:76:95:0a:0f:d8:df:d5:bd:12:
1f:35:2f:28:17:6c:d2:98:c1:a8:09:64:77:6e:47:
37:ba:ce:ac:59:5e:68:9d:7f:72:d6:89:c5:06:41:
29:3e:59:3e:dd:26:f5:24:c9:11:a7:5a:a3:4c:40:
1f:46:a1:99:b5:a7:3a:51:6e:86:3b:9e:7d:72:a7:
12:05:78:59:ed:3e:51:78:15:0b:03:8f:8d:d0:2f:
05:b2:3e:7b:4a:1c:4b:73:05:12:fc:c6:ea:e0:50:
13:7c:43:93:74:b3:ca:74:e7:8e:1f:01:08:d0:30:
d4:5b:71:36:b4:07:ba:c1:30:30:5c:48:b7:82:3b:
98:a6:7d:60:8a:a2:a3:29:82:cc:ba:bd:83:04:1b:
a2:83:03:41:a1:d6:05:f1:1b:c2:b6:f0:a8:7c:86:
3b:46:a8:48:2a:88:dc:76:9a:76:bf:1f:6a:a5:3d:
19:8f:eb:38:f3:64:de:c8:2b:0d:0a:28:ff:f7:db:
e2:15:42:d4:22:d0:27:5d:e1:79:fe:18:e7:70:88:
ad:4e:e6:d9:8b:3a:c6:dd:27:51:6e:ff:bc:64:f5:
33:43:4f

182
SARA/SSL/old/full_test_HTTPS.py Normal file
View File

@@ -0,0 +1,182 @@
'''
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/full_test_HTTPS.py ttyAMA2 aircarto.fr
To do: need to add profile id as parameter
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
#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 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("****")
print("Add certificate")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.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'
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)
ser_sara.write(certificate)
response_SARA_2 = read_complete_response(ser_sara)
print("Write certificate data")
print(response_SARA_2)
time.sleep(1)
#step 2: set the security profile 2 to trusted root
print("****")
print("SET security profile")
command = f'AT+USECPRF=2,0,1\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_3 = read_complete_response(ser_sara)
print(response_SARA_3)
time.sleep(1)
#step 3: set the security profile 2 to the imported certificate
print("****")
print("SET certificate")
command = f'AT+USECPRF=2,3,"myCertificate2"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_4 = read_complete_response(ser_sara)
print(response_SARA_4)
time.sleep(1)
#step 4: set url
print("****")
print("SET URL")
command = f'AT+UHTTP=1,1,"{url}"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
#step 4: set url
print("****")
print("SET PORT")
command = f'AT+UHTTP=1,5,443\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
print(response_SARA_55)
time.sleep(1)
#step 4: set url to SSL
print("****")
print("SET SSL")
command = f'AT+UHTTP=1,6,1,2\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
#step 4: trigger the request
print("****")
print("Trigger POST REQUEST")
command = f'AT+UHTTPC=1,1,"/tests/test.php","https.resp"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_6 = read_complete_response(ser_sara)
print(response_SARA_6)
time.sleep(1)
#READ REPLY
print("****")
print("Read reply from server")
ser_sara.write(b'AT+URDFILE="https.resp"\r')
response_SARA_7 = read_complete_response(ser_sara)
print("Reply from server:")
print(response_SARA_7)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

83
SARA/SSL/old/sara_add_certif.py Normal file
View File

@@ -0,0 +1,83 @@
'''
Script to add the SSL certificate
ex:
python3 /var/www/nebuleair_pro_4g/SARA/SSL/sara_add_certif.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
#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
)
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
return response.decode('utf-8')
try:
certificate='00:ad:e8:24:73:f4:14:37:f3:9b:9e:2b:57:28:1c:87:be:dc:b7:df:38:90:8c:6e:3c:e6:57:a0:78:f7:75:c2:a2:fe:f5:6a:6e:f6:00:4f:28:db:de:68:86:6c:44:93:b6:b1:63:fd:14:12:6b:bf:1f:d2:ea:31:9b:21:7e:d1:33:3c:ba:48:f5:dd:79:df:b3:b8:ff:12:f1:21:9a:4b:c1:8a:86:71:69:4a:66:66:6c:8f:7e:3c:70:bf:ad:29:22:06:f3:e4:c0:e6:80:ae:e2:4b:8f:b7:99:7e:94:03:9f:d3:47:97:7c:99:48:23:53:e8:38:ae:4f:0a:6f:83:2e:d1:49:57:8c:80:74:b6:da:2f:d0:38:8d:7b:03:70:21:1b:75:f2:30:3c:fa:8f:ae:dd:da:63:ab:eb:16:4f:c2:8e:11:4b:7e:cf:0b:e8:ff:b5:77:2e:f4:b2:7b:4a:e0:4c:12:25:0c:70:8d:03:29:a0:e1:53:24:ec:13:d9:ee:19:bf:10:b3:4a:8c:3f:89:a3:61:51:de:ac:87:07:94:f4:63:71:ec:2e:e2:6f:5b:98:81:e1:89:5c:34:79:6c:76:ef:3b:90:62:79:e6:db:a4:9a:2f:26:c5:d0:10:e1:0e:de:d9:10:8e:16:fb:b7:f7:a8:f7:c7:e5:02:07:98:8f:36:08:95:e7:e2:37:96:0d:36:75:9e:fb:0e:72:b1:1d:9b:bc:03:f9:49:05:d8:81:dd:05:b4:2a:d6:41:e9:ac:01:76:95:0a:0f:d8:df:d5:bd:12:1f:35:2f:28:17:6c:d2:98:c1:a8:09:64:77:6e:47:37:ba:ce:ac:59:5e:68:9d:7f:72:d6:89:c5:06:41:29:3e:59:3e:dd:26:f5:24:c9:11:a7:5a:a3:4c:40:1f:46:a1:99:b5:a7:3a:51:6e:86:3b:9e:7d:72:a7:12:05:78:59:ed:3e:51:78:15:0b:03:8f:8d:d0:2f:05:b2:3e:7b:4a:1c:4b:73:05:12:fc:c6:ea:e0:50:13:7c:43:93:74:b3:ca:74:e7:8e:1f:01:08:d0:30:d4:5b:71:36:b4:07:ba:c1:30:30:5c:48:b7:82:3b:98:a6:7d:60:8a:a2:a3:29:82:cc:ba:bd:83:04:1b:a2:83:03:41:a1:d6:05:f1:1b:c2:b6:f0:a8:7c:86:3b:46:a8:48:2a:88:dc:76:9a:76:bf:1f:6a:a5:3d:19:8f:eb:38:f3:64:de:c8:2b:0d:0a:28:ff:f7:db:e2:15:42:d4:22:d0:27:5d:e1:79:fe:18:e7:70:88:ad:4e:e6:d9:8b:3a:c6:dd:27:51:6e:ff:bc:64:f5:33:43:4f'
size_of_string = len(certificate)
command = f'AT+USECMNG=0,0,"myCertificate",{size_of_string}\r'
ser.write((command + '\r').encode('utf-8'))
response_SARA_1 = read_complete_response(ser)
print("Write certificate")
print(response_SARA_1)
time.sleep(1)
ser.write(certificate.encode())
response_SARA_2 = read_complete_response(ser)
print("Write to memory:")
print(response_SARA_2)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

74
SARA/SSL/old/sara_read_certif.py Normal file
View File

@@ -0,0 +1,74 @@
'''
Script to add the SSL certificate
ex:
python3 /var/www/nebuleair_pro_4g/SARA/SSL/sara_read_certif.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
#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
)
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
return response.decode('utf-8')
ser.write(b'AT+USECMNG=3')
try:
response2 = read_complete_response(ser)
print("Response:")
print(response2)
print("<----")
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser.is_open:
ser.close()
#print("Serial closed")

5
SARA/SSL/open_ssl_script.sh Normal file
View File

@@ -0,0 +1,5 @@
#!/bin/bash
openssl s_client -connect aircarto.fr:443 -cipher TLS_AES_256_GCM_SHA384
openssl s_client -connect aircarto.fr:443 -tls1_3 -ciphersuites TLS_AES_256_GCM_SHA384
openssl s_client -connect api-prod.uspot.probesys.net:443 -tls1_2 -ciphersuites TLS_AES_256_GCM_SHA384 -CAfile /var/www/nebuleair_pro_4g/SARA/SSL/certificate/isrgrootx1.pem
openssl s_client -connect aircarto.fr:443 -tls1_2 -ciphersuites TLS_AES_256_GCM_SHA384 -CAfile /var/www/nebuleair_pro_4g/SARA/SSL/certificate/isrgrootx1.pem

149
SARA/SSL/prepareUspotProfile.py Normal file
View File

@@ -0,0 +1,149 @@
'''
Script to set the URL for a HTTP request
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/prepareUspotProfile.py ttyAMA2 api-prod.uspot.probesys.net
To do: need to add profile id as parameter
First profile id:
AT+UHTTP=0,1,"data.nebuleair.fr"
Second profile id:
AT+UHTTP=1,1,"api-prod.uspot.probesys.net"
'''
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)
send_uSpot = config.get('send_uSpot', False)
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
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("****")
print("Add certificate")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.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'
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)
ser_sara.write(certificate)
response_SARA_2 = read_complete_response(ser_sara)
print("Write certificate data")
print(response_SARA_2)
time.sleep(1)
#step 2: set the security profile 2 to trusted root
print("****")
print("SET security profile")
command = f'AT+USECPRF=2,0,1\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_3 = read_complete_response(ser_sara)
print(response_SARA_3)
time.sleep(1)
#step 3: set the security profile 2 to the imported certificate
print("****")
print("SET certificate")
command = f'AT+USECPRF=2,3,"myCertificate2"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_4 = read_complete_response(ser_sara)
print(response_SARA_4)
time.sleep(1)
#step 4: set url
print("****")
print("SET URL")
command = f'AT+UHTTP=1,1,"{url}"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
#step 4: set url
print("****")
print("SET PORT")
command = f'AT+UHTTP=1,5,443\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
print(response_SARA_55)
time.sleep(1)
#step 4: set url to SSL
print("****")
print("SET SSL")
command = f'AT+UHTTP=1,6,1,2\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
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/prepareUspotProfile_V2.py Normal file
View File

@@ -0,0 +1,133 @@
'''
Script to set the URL for a HTTP request
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/prepareUspotProfile_V2.py ttyAMA2 api-prod.uspot.probesys.net
To do: need to add profile id as parameter
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
#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 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
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("****")
print("Add certificate")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.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'
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)
ser_sara.write(certificate)
response_SARA_2 = read_complete_response(ser_sara)
print("Write certificate data")
print(response_SARA_2)
time.sleep(1)
#step 4: set url
print("****")
print("SET URL")
command = f'AT+UHTTP=1,1,"{url}"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
#step 4: set PORT
print("****")
print("SET PORT")
command = f'AT+UHTTP=1,5,443\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_55 = read_complete_response(ser_sara)
print(response_SARA_55)
time.sleep(1)
#step 4: set url to SSL
print("****")
print("SET SSL")
command = f'AT+UHTTP=1,6,1,2\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5 = read_complete_response(ser_sara)
print(response_SARA_5)
time.sleep(1)
except serial.SerialException as e:
print(f"Error: {e}")
finally:
if ser_sara.is_open:
ser_sara.close()
print("****")
#print("Serial closed")

45
SARA/SSL/request.py Normal file
View File

@@ -0,0 +1,45 @@
'''
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/request.py
'''
import requests
import logging
import http.client as http_client
# Enable HTTP and HTTPS verbose logging
http_client.HTTPConnection.debuglevel = 1
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("http.client").setLevel(logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)
logging.getLogger("requests").setLevel(logging.DEBUG)
# Suppress logging from unrelated libraries
logging.getLogger("chardet").setLevel(logging.INFO)
logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
url_microSpot = "https://api-prod.uspot.probesys.net/nebuleair?token=2AFF6dQk68daFZ"
url_aircarto = "https://aircarto.fr/tests/test.php"
payload = {
"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"}
]
}
cert_path = "/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.pem"
try:
response = requests.post(url_aircarto, json=payload, verify=cert_path)
print("Response Status Code:", response.status_code)
print("Response Text:", response.text)
except requests.exceptions.RequestException as e:
print("An error occurred:", e)

75
SARA/SSL/sara_add_certif_v2.py Normal file
View File

@@ -0,0 +1,75 @@
'''
Script to add the SSL certificate
ex:
python3 /var/www/nebuleair_pro_4g/SARA/SSL/sara_add_certif_v2.py ttyAMA2 2
'''
import serial
import time
import sys
import json
import os
parameter = sys.argv[1:] # Exclude the script name
port = '/dev/' + parameter[0] # e.g., ttyAMA2
timeout = float(parameter[1]) # e.g., 2 seconds
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'
config = load_config(config_file)
baudrate = config.get('SaraR4_baudrate', 115200)
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)
return response.decode('utf-8', errors='replace')
try:
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.der", "rb") as cert_file:
certificate = cert_file.read()
size_of_string = len(certificate)
ser = serial.Serial(port=port, baudrate=baudrate, timeout=timeout)
command = f'AT+USECMNG=0,0,"myCertificate2",{size_of_string}\r'
ser.write((command + '\r').encode('utf-8'))
response_SARA_1 = read_complete_response(ser)
print("Write certificate metadata")
print(response_SARA_1)
time.sleep(1)
ser.write(certificate)
response_SARA_2 = read_complete_response(ser)
print("Write certificate data")
print(response_SARA_2)
except serial.SerialException as e:
print(f"Serial Error: {e}")
except FileNotFoundError as e:
print(f"File Error: {e}")
except Exception as e:
print(f"Unexpected Error: {e}")
finally:
if 'ser' in locals() and ser.is_open:
ser.close()

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")