Mathieu MoranandCopilot
1b39ac0cc8
- app_alerts.py: null guard on list_local_videos() result (.get('files', []))
- app_cellular.py: remove erroneous .encode(errors='replace') on str response body
- app_settings.py: initialize apn='' when isinstance check fails (NameError guard)
- app_upload.py: move json import to top of file instead of inside except handler
- server/app.py: wrap int(request.args.get('limit')) in try/except ValueError
- server/app.py: wrap os.path.commonpath() in try/except ValueError (cross-root paths)
- tests/test_app_settings.py: add missing trips_upload_enabled/video_convert_enabled
to valid_payload() (fields added in Phase 23.5 but test not updated)
All 63 tests pass.
Co-authored-by: Copilot <[email protected]>
250 lines
9.1 KiB
Python
250 lines
9.1 KiB
Python
"""SIM868 GSM/GPRS cellular fallback for uploading trip/GPS tracking batches
|
|
when there is no WiFi/Ethernet path to the remote server.
|
|
|
|
*** UNTESTED - no SIM card was available to validate this against a real
|
|
*** cellular network. Written defensively (every AT exchange is wrapped so a
|
|
*** failure just raises CellularError, never crashes the app) but the exact
|
|
*** AT command dialect/timing may need small tweaks once a SIM is inserted
|
|
*** and this can actually be exercised. See "Verifying on real hardware"
|
|
*** below and the Phase 9 notes in PLAN.md.
|
|
|
|
Why AT-command HTTP instead of a real ppp0 network interface: the SIM868's
|
|
GNSS (used by modules/app_gps.py) and its GSM/GPRS radio are both driven over
|
|
the *same* physical UART (/dev/serial0). A classic pppd/chat dial-up would put
|
|
that UART into raw PPP framing for as long as the link is up, which would
|
|
starve/conflict with the GNSS AT+CGNSINF polling already running continuously
|
|
in app_gps.py. SIMCOM modules instead expose a built-in "HTTP(S) Application"
|
|
AT command set (AT+HTTPINIT / AT+HTTPPARA / AT+HTTPACTION / AT+HTTPREAD) that
|
|
stays in normal AT command mode the whole time, so it can safely interleave
|
|
with GNSS polling through the same serial lock (modules.app_gps._serial_lock) -
|
|
each exchange just briefly delays the next GPS poll, it never takes over the
|
|
port.
|
|
|
|
This only carries small JSON tracking-batch payloads - dashcam video uploads
|
|
remain WiFi/Ethernet-only (see modules/app_upload.py), both because cellular
|
|
data is far too slow/limited for video files and because the modem's HTTP
|
|
AT interface has a small internal buffer unsuited to multi-MB bodies.
|
|
"""
|
|
import os
|
|
import time
|
|
|
|
from modules import app_gps, app_sql
|
|
|
|
GPRS_BRINGUP_TIMEOUT = 30 # seconds to wait for AT+CIICR / AT+CIFSR
|
|
HTTP_ACTION_TIMEOUT = 60 # seconds to wait for the +HTTPACTION URC (2G/3G can be slow)
|
|
HTTP_USERDATA_MAX_BYTES = 1024
|
|
|
|
_gprs_context_up = False
|
|
|
|
|
|
class CellularError(Exception):
|
|
"""Raised for any cellular-modem/AT-command failure - always caught by
|
|
the caller (modules/app_upload.py), never allowed to crash the app."""
|
|
|
|
|
|
def is_configured():
|
|
"""True once the user has filled in an APN and flipped the enabled flag
|
|
in the Settings page or legacy environment configuration. Off by default -
|
|
nothing cellular-related runs unless explicitly opted into."""
|
|
enabled = app_sql.get_setting(
|
|
"cellular_enabled",
|
|
"1" if os.getenv("YARIS_CELLULAR_ENABLED", "") == "1" else "0",
|
|
)
|
|
return enabled == "1" and bool(get_apn())
|
|
|
|
|
|
def get_apn():
|
|
return app_sql.get_setting(
|
|
"cellular_apn", os.getenv("YARIS_CELLULAR_APN", "")
|
|
)
|
|
|
|
|
|
def _read_until(ser, terminators, timeout):
|
|
"""Read from the modem until one of the given substrings shows up in the
|
|
accumulated buffer, or timeout. Returns the buffer (decoded, best-effort)."""
|
|
deadline = time.time() + timeout
|
|
buf = ""
|
|
while time.time() < deadline:
|
|
chunk = ser.read(1024)
|
|
if chunk:
|
|
buf += chunk.decode(errors="replace")
|
|
if any(t in buf for t in terminators):
|
|
return buf
|
|
else:
|
|
time.sleep(0.1)
|
|
return buf
|
|
|
|
|
|
def _at(ser, cmd, terminators=("OK", "ERROR"), timeout=5):
|
|
"""Send one AT command and wait for a terminator. Raises CellularError
|
|
if the modem responds with ERROR or nothing recognizable within timeout."""
|
|
ser.reset_input_buffer()
|
|
ser.write((cmd + "\r\n").encode())
|
|
resp = _read_until(ser, terminators, timeout)
|
|
if "ERROR" in resp and "OK" not in resp:
|
|
raise CellularError(f"{cmd!r} -> {resp!r}")
|
|
if not any(t in resp for t in terminators):
|
|
raise CellularError(f"{cmd!r} timed out, got {resp!r}")
|
|
return resp
|
|
|
|
|
|
def _ensure_gprs_context():
|
|
"""Bring up the GPRS data context (APN attach) if not already up.
|
|
|
|
Reuses app_gps's serial connection/lock - see module docstring for why.
|
|
"""
|
|
global _gprs_context_up
|
|
|
|
with app_gps._serial_lock:
|
|
ser = app_gps._get_serial()
|
|
|
|
if _gprs_context_up:
|
|
# Cheap check that the context is still alive before trusting it.
|
|
try:
|
|
resp = _at(ser, "AT+CIFSR", terminators=("\r\n",), timeout=5)
|
|
if "." in resp and "ERROR" not in resp:
|
|
return
|
|
except CellularError:
|
|
pass
|
|
_gprs_context_up = False
|
|
|
|
_at(ser, "AT+CGATT=1", timeout=10)
|
|
|
|
apn = _validate_http_parameter(get_apn(), "APN")
|
|
apn_user = _validate_http_parameter(
|
|
os.getenv("YARIS_CELLULAR_APN_USER", ""),
|
|
"APN username",
|
|
)
|
|
apn_password = _validate_http_parameter(
|
|
os.getenv("YARIS_CELLULAR_APN_PASS", ""),
|
|
"APN password",
|
|
)
|
|
apn_cmd = f'AT+CSTT="{apn}","{apn_user}","{apn_password}"'
|
|
_at(ser, apn_cmd, timeout=10)
|
|
|
|
_at(ser, "AT+CIICR", timeout=GPRS_BRINGUP_TIMEOUT)
|
|
|
|
resp = _at(ser, "AT+CIFSR", terminators=("\r\n",), timeout=10)
|
|
if "ERROR" in resp or "." not in resp:
|
|
raise CellularError(f"AT+CIFSR did not return an IP: {resp!r}")
|
|
|
|
_gprs_context_up = True
|
|
|
|
|
|
def _validate_http_parameter(value, label):
|
|
if not isinstance(value, str) or any(
|
|
char in value for char in ('"', "\r", "\n")
|
|
):
|
|
raise CellularError(f"{label} contains unsupported characters")
|
|
return value
|
|
|
|
|
|
def http_post(
|
|
url,
|
|
body_bytes,
|
|
extra_headers,
|
|
content_type="application/json",
|
|
timeout=HTTP_ACTION_TIMEOUT,
|
|
):
|
|
"""POST a small body to `url` via the modem's built-in HTTP client.
|
|
|
|
Returns (status_code, response_body_bytes). Raises CellularError on any
|
|
failure (modem not responding, no network attach, HTTP-layer failure,
|
|
etc.) - callers should catch this broadly and treat it the same as a
|
|
network-unreachable condition.
|
|
"""
|
|
if not is_configured():
|
|
raise CellularError("cellular fallback not configured (APN/enabled flag not set)")
|
|
_validate_http_parameter(url, "URL")
|
|
_validate_http_parameter(content_type, "Content type")
|
|
for name, value in extra_headers.items():
|
|
_validate_http_parameter(name, "Header name")
|
|
_validate_http_parameter(value, f"{name} header")
|
|
header_lines = r"\r\n".join(
|
|
f"{name}: {value}" for name, value in extra_headers.items()
|
|
)
|
|
if header_lines:
|
|
header_lines += r"\r\n"
|
|
if len(header_lines.encode("utf-8")) > HTTP_USERDATA_MAX_BYTES:
|
|
raise CellularError("HTTP headers exceed the modem's 1024-byte limit")
|
|
|
|
_ensure_gprs_context()
|
|
|
|
with app_gps._serial_lock:
|
|
ser = app_gps._get_serial()
|
|
try:
|
|
_at(ser, "AT+HTTPTERM", timeout=3)
|
|
except CellularError:
|
|
pass # fine if nothing was initialized yet
|
|
|
|
_at(ser, "AT+HTTPINIT", timeout=5)
|
|
try:
|
|
_at(ser, 'AT+HTTPPARA="CID",1', timeout=5)
|
|
_at(ser, f'AT+HTTPPARA="URL","{url}"', timeout=5)
|
|
if url.lower().startswith("https"):
|
|
try:
|
|
_at(ser, "AT+HTTPSSL=1", timeout=5)
|
|
except CellularError:
|
|
# Older SIMCOM firmware/certificate stores sometimes
|
|
# don't support this - surfacing as a clear error is more
|
|
# useful than silently falling back to plain HTTP against
|
|
# an HTTPS-only endpoint.
|
|
raise CellularError(
|
|
"modem does not support AT+HTTPSSL - HTTPS upload via "
|
|
"cellular is unavailable on this firmware. Consider a "
|
|
"plain-HTTP router port-forward path for cellular use."
|
|
)
|
|
_at(ser, f'AT+HTTPPARA="CONTENT","{content_type}"', timeout=5)
|
|
|
|
if header_lines:
|
|
_at(
|
|
ser,
|
|
f'AT+HTTPPARA="USERDATA","{header_lines}"',
|
|
timeout=5,
|
|
)
|
|
|
|
_at(
|
|
ser,
|
|
f"AT+HTTPDATA={len(body_bytes)},{int(timeout * 1000)}",
|
|
terminators=("DOWNLOAD", "ERROR"),
|
|
timeout=5,
|
|
)
|
|
ser.write(body_bytes)
|
|
_read_until(ser, ("OK", "ERROR"), timeout=10)
|
|
|
|
action_resp = _at(
|
|
ser,
|
|
"AT+HTTPACTION=1",
|
|
terminators=("+HTTPACTION:",),
|
|
timeout=timeout,
|
|
)
|
|
|
|
status_code = 0
|
|
for line in action_resp.splitlines():
|
|
if "+HTTPACTION:" in line:
|
|
# +HTTPACTION: <method>,<status_code>,<data_len>
|
|
parts = line.split(":", 1)[1].strip().split(",")
|
|
if len(parts) >= 2:
|
|
status_code = int(parts[1])
|
|
break
|
|
|
|
read_resp = _at(ser, "AT+HTTPREAD", terminators=("OK", "ERROR"), timeout=15)
|
|
body = read_resp
|
|
|
|
return status_code, body
|
|
finally:
|
|
try:
|
|
_at(ser, "AT+HTTPTERM", timeout=5)
|
|
except CellularError:
|
|
pass
|
|
|
|
|
|
def http_post_json(url, body_bytes, extra_headers, timeout=HTTP_ACTION_TIMEOUT):
|
|
return http_post(
|
|
url,
|
|
body_bytes,
|
|
extra_headers,
|
|
content_type="application/json",
|
|
timeout=timeout,
|
|
)
|