"""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: ,, 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.encode(errors="replace") 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, )