Page modem: section tests Miotiq UDP + masquer tests HTTP si SIM Miotiq

- Nouveau script SARA/sara_test_udp.py (test socket UDP vers 192.168.0.20:4242)
- Section "Tests Miotiq (UDP)" avec PSD setup, test socket, placeholder aller-retour
- Masque les tests HTTP/Send message quand send_miotiq est actif
- Endpoint launcher.php sara_test_udp

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
PaulVua
2026-04-27 15:54:39 +02:00
parent a68af89612
commit b8b70a5a54
3 changed files with 257 additions and 0 deletions

154
SARA/sara_test_udp.py Normal file
View File

@@ -0,0 +1,154 @@
r'''
____ _ ____ _
/ ___| / \ | _ \ / \
\___ \ / _ \ | |_) | / _ \
___) / ___ \| _ < / ___ \
|____/_/ \_\_| \_\/_/ \_\
Test UDP socket connectivity to Miotiq private network (192.168.0.20:4242)
Creates a UDP socket, connects, writes a small test payload, and closes.
Each AT step is verified for errors.
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/sara_test_udp.py
'''
import serial
import time
import sys
import re
ser_sara = serial.Serial(
port='/dev/ttyAMA2',
baudrate=115200,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
timeout=2
)
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2, wait_for_lines=None, debug=False):
if wait_for_lines is None:
wait_for_lines = []
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
start_time = time.time()
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
decoded_response = response.decode('utf-8', errors='replace')
for target_line in wait_for_lines:
if target_line in decoded_response:
return decoded_response
elif time.time() > end_time:
break
time.sleep(0.1)
return response.decode('utf-8', errors='replace')
# Result tracking
steps = []
def log_step(name, success, detail=""):
steps.append({"name": name, "success": success, "detail": detail})
icon = "" if success else ""
print(f'{icon} {name}')
if detail:
print(f'<small class="text-muted">{detail}</small>')
try:
sys.stdout.reconfigure(line_buffering=True)
ser_sara.reset_input_buffer()
# Step 1: Create UDP socket (protocol 17 = UDP)
command = 'AT+USOCR=17\r'
ser_sara.write(command.encode('utf-8'))
response = read_complete_response(ser_sara, wait_for_lines=["OK", "+CME ERROR", "ERROR"])
if "+CME ERROR" in response or "ERROR" in response:
log_step("Création socket UDP", False, "AT+USOCR=17 → erreur. La connexion PDP n'est peut-être pas établie.")
print('<br><strong>Suggestion :</strong> Lancez "Setup PSD connection" puis réessayez.')
sys.exit(1)
# Extract socket ID
match = re.search(r'\+USOCR:\s*(\d+)', response)
if not match:
log_step("Création socket UDP", False, "Impossible d'extraire le socket ID")
sys.exit(1)
socket_id = match.group(1)
log_step("Création socket UDP", True, f"Socket ID: {socket_id}")
# Step 2: Connect to Miotiq server
ser_sara.reset_input_buffer()
command = f'AT+USOCO={socket_id},"192.168.0.20",4242\r'
ser_sara.write(command.encode('utf-8'))
response = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=5, wait_for_lines=["OK", "+CME ERROR", "ERROR"])
if "+CME ERROR" in response or "ERROR" in response:
log_step("Connexion 192.168.0.20:4242", False, f"Erreur connexion au serveur Miotiq")
# Close socket
ser_sara.write(f'AT+USOCL={socket_id}\r'.encode('utf-8'))
read_complete_response(ser_sara, wait_for_lines=["OK", "ERROR"], timeout=1, end_of_response_timeout=1)
sys.exit(1)
log_step("Connexion 192.168.0.20:4242", True)
# Step 3: Write test payload (4 bytes: "TEST")
test_payload = b'TEST'
ser_sara.reset_input_buffer()
command = f'AT+USOWR={socket_id},{len(test_payload)}\r'
ser_sara.write(command.encode('utf-8'))
response = read_complete_response(ser_sara, wait_for_lines=["@", "OK", "+CME ERROR", "ERROR"])
if "@" not in response:
log_step("Écriture payload test", False, "Le modem n'a pas envoyé le prompt @")
ser_sara.reset_input_buffer()
ser_sara.write(f'AT+USOCL={socket_id}\r'.encode('utf-8'))
read_complete_response(ser_sara, wait_for_lines=["OK", "ERROR"], timeout=1, end_of_response_timeout=1)
sys.exit(1)
# Send the actual bytes
ser_sara.write(test_payload)
response = read_complete_response(ser_sara, wait_for_lines=["OK", "+CME ERROR", "ERROR"])
if "+CME ERROR" in response or "ERROR" in response:
log_step("Écriture payload test", False, "Erreur lors de l'envoi des données")
ser_sara.write(f'AT+USOCL={socket_id}\r'.encode('utf-8'))
read_complete_response(ser_sara, wait_for_lines=["OK", "ERROR"], timeout=1, end_of_response_timeout=1)
sys.exit(1)
log_step("Écriture payload test", True, f"{len(test_payload)} bytes envoyés")
# Step 4: Close socket
ser_sara.reset_input_buffer()
command = f'AT+USOCL={socket_id}\r'
ser_sara.write(command.encode('utf-8'))
response = read_complete_response(ser_sara, wait_for_lines=["OK", "+CME ERROR", "ERROR"])
if "+CME ERROR" in response or "ERROR" in response:
log_step("Fermeture socket", False)
else:
log_step("Fermeture socket", True)
# Summary
all_ok = all(s["success"] for s in steps)
if all_ok:
print('<br><strong class="text-success">Toutes les étapes OK — le modem peut envoyer des données UDP vers Miotiq.</strong>')
else:
print('<br><strong class="text-danger">Certaines étapes ont échoué.</strong>')
except serial.SerialException as e:
print(f'❌ Erreur série: {e}')
except Exception as e:
print(f'❌ Erreur: {e}')
finally:
if ser_sara.is_open:
ser_sara.close()

View File

@@ -389,6 +389,12 @@ if ($type == "sara_psd_setup") {
echo $output;
}
if ($type == "sara_test_udp") {
$command = 'sudo /usr/bin/python3 -u /var/www/nebuleair_pro_4g/SARA/sara_test_udp.py';
$output = shell_exec($command);
echo $output;
}
if ($type == "git_pull") {
$command = 'sudo git pull';

View File

@@ -302,6 +302,48 @@
</div>
</div>
-->
<div id="section_miotiq_tests" style="display:none;">
<h3>Tests Miotiq (UDP)</h3>
<div class="row mb-3">
<div class="col-sm-4">
<div class="card">
<div class="card-body">
<p class="card-text"><strong>1. Setup PSD connection</strong></p>
<p class="text-muted small mb-2">Active la connexion PDP nécessaire pour les sockets UDP.</p>
<button class="btn btn-primary" onclick="PSD_setup_miotiq()">Start</button>
<div id="loading_PSD_miotiq" class="spinner-border spinner-border-sm" style="display: none;" role="status"></div>
<div id="response_psd_setup_miotiq"></div>
</div>
</div>
</div>
<div class="col-sm-4">
<div class="card">
<div class="card-body">
<p class="card-text"><strong>2. Test socket UDP</strong></p>
<p class="text-muted small mb-2">Crée un socket UDP, se connecte à 192.168.0.20:4242, envoie un payload test, puis ferme le socket.</p>
<button class="btn btn-primary" onclick="testUdpSocket()">Test UDP</button>
<div id="loading_udp_test" class="spinner-border spinner-border-sm" style="display: none;" role="status"></div>
<div id="response_udp_test"></div>
</div>
</div>
</div>
<div class="col-sm-4">
<div class="card">
<div class="card-body">
<p class="card-text"><strong>3. Test aller-retour</strong></p>
<p class="text-muted small mb-2">Envoie un payload spécial puis vérifie la réception via l'API descendante Miotiq. (à venir)</p>
<button class="btn btn-secondary" disabled>Bientôt disponible</button>
</div>
</div>
</div>
</div>
</div>
<div id="section_http_tests">
<h3>Test HTTP server comm.</h3>
<div class="row mb-3">
<!-- SET URL -->
@@ -397,6 +439,7 @@
</div>
</div><!-- end section_http_tests -->
<!-- toast -->
@@ -480,6 +523,12 @@ document.addEventListener('DOMContentLoaded', function () {
// Show/hide config mode alert banner
const alertEl = document.getElementById("configmode_alert");
if (alertEl) alertEl.style.display = response.modem_config_mode ? "flex" : "none";
// Show Miotiq tests OR HTTP tests depending on SIM type
const sectionHttp = document.getElementById("section_http_tests");
const sectionMiotiq = document.getElementById("section_miotiq_tests");
if (sectionHttp) sectionHttp.style.display = response.send_miotiq ? "none" : "";
if (sectionMiotiq) sectionMiotiq.style.display = response.send_miotiq ? "" : "none";
} else {
console.error("Checkbox element not found");
}
@@ -1626,6 +1675,54 @@ function update_modem_configMode(param, checked){
function PSD_setup_miotiq() {
console.log("Setup PSD connection (Miotiq):");
$("#loading_PSD_miotiq").show();
$("#response_psd_setup_miotiq").empty();
$.ajax({
url: 'launcher.php?type=sara_psd_setup',
dataType: 'text',
method: 'GET',
success: function(response) {
console.log(response);
$("#loading_PSD_miotiq").hide();
const formattedResponse = response.replace(/\n/g, "<br>");
$("#response_psd_setup_miotiq").html(formattedResponse);
},
error: function(xhr, status, error) {
console.error('AJAX request failed:', status, error);
$("#loading_PSD_miotiq").hide();
}
});
}
function testUdpSocket() {
console.log("Test UDP socket to Miotiq:");
$("#response_udp_test").empty();
$("#loading_udp_test").show();
$.ajax({
url: 'launcher.php?type=sara_test_udp',
dataType: 'text',
method: 'GET',
timeout: 30000,
success: function(response) {
console.log(response);
$("#loading_udp_test").hide();
const formattedResponse = response.replace(/\n/g, "<br>");
$("#response_udp_test").html(formattedResponse);
},
error: function(xhr, status, error) {
console.error('AJAX request failed:', status, error);
$("#loading_udp_test").hide();
$("#response_udp_test").html(`
<div class="alert alert-danger py-2 mt-2">
<strong>Erreur de communication</strong><br>
<small>${error}</small>
</div>`);
}
});
}
// Self test functions are now in assets/js/selftest.js