"""Quectel EG915 GSM/LTE modem driver.
The :class:`eg915` class below is **verbatim** from the device firmware
(``firmware_core.py`` of Logger Pro and the findyIoT module) - same signatures, same bodies -
with documentation added. The pin constants and ``apn`` mirror the
firmware module globals the class reads.
All hardware access happens inside methods; importing this module
performs no I/O (``machine``/``utime`` are mocked for this build).
"""
import time
import utime
import gc
import machine
#: Access Point Name used for the PDP context.
apn = "m2m.tele2.com"
#: Modem power rail pin.
P_Power_GSM = 17
#: Modem PWRKEY pin.
P_PWRKey = 23
#: ESP32 pin wired to the modem TX line.
P_TxGSM = 36
#: ESP32 pin wired to the modem RX line.
P_RXGSM = 21
#: Modem DTR pin (slow-clock sleep control).
P_DTR = 5
#: Modem PSM status input (unused by the driver).
P_PSM_IN = 35
#: Modem PON trigger (unused by this board revision).
P_PON_TRIG = 16
# Runtime globals stamped by the driver (module-level on the device).
registrationStart = 0
registrationTime = 0
noReg = 0
[docs]
class eg915(object):
"""Driver for the Quectel EG915 modem on UART1.
Verbatim class from ``firmware_core.py``. The constructor takes no
arguments - the UART/control pins come from the module-level pin
constants and ``apn``, exactly as on the device.
Note:
This board has no dedicated RESET_GSM/PSM pins, so
:meth:`reset` power-cycles the supply rail instead of pulsing a
reset line, and there are no RAT-limiting or PSM helpers.
"""
[docs]
def __init__(self):
"""Prepares the modem control pins in a safe state.
Reads the pin numbers and APN from the module constants
(``P_TxGSM``, ``P_RXGSM``, ``P_DTR``, ``P_PWRKey``,
``P_Power_GSM``, ``apn``). No UART is created yet - that
happens lazily in :meth:`isOn`.
"""
global P_TxGSM
global P_RXGSM
global P_DTR
global P_PWRKey
global P_Power_GSM
#,power,key,tx,rx,br
self.tx = P_RXGSM
self.rx = P_TxGSM
self.br = 115200
self.apn = apn
self.pwr = machine.Pin(P_Power_GSM,machine.Pin.OUT)
self.dtr = machine.Pin(P_DTR, machine.Pin.OUT)
self.key = machine.Pin(P_PWRKey,machine.Pin.OUT)#.value(0)
self.key.value(0)
self.dtr.value(0)
return
[docs]
def isOn(self):
"""Checks whether the modem answers AT commands.
(Re)creates UART1 (falling back to smaller buffers when RAM is
low), sends ``AT`` and looks for ``OK``. On success also reads
the SIM ICCID as a side effect.
Returns:
bool: True if the modem replied ``OK``.
"""
try:
self.uart = machine.UART(1,self.br, rx=self.rx, tx=self.tx, txbuf=1024,rxbuf=2048)
except:
try:
gc.collect()
self.uart = machine.UART(1,self.br, rx=self.rx, tx=self.tx, txbuf=512,rxbuf=512)
except:
return False
#print("Modem Status ",self.status.value())
self.us("AT")
at = self.us("AT")
if type(at) != bytes and type(at) != str:
return False
#=1,"IP","CMNET"
#self.us('AT+CGDCONT=0,"IP","'+self.apn+'"')
if "OK" in at:
self.us('AT+QCCID')
return True
else :
return False
[docs]
def turnOn(self):
"""Powers the modem on and waits until it accepts AT commands.
Raises the power rail (held through deep sleep), waits 1 s for
VBAT to stabilize, wakes the modem from slow-clock mode, then
pulses PWRKEY - up to 3 attempts, re-pulsing when the modem
stays silent (a cold modem can ignore the first pulse). Stamps
the global ``registrationStart`` used by :meth:`registerNetwork`.
Returns:
bool: Always True; check :meth:`isOn` for the actual state.
"""
global registrationStart
#self.pwr.value(1)
#self.key.value(1)
if self.isOn() == False:
if self.pwr.value() == 0:
print("!!!! Modem was OFF !!!!")
self.pwr = machine.Pin(P_Power_GSM,machine.Pin.OUT,hold=False)
self.pwr.value(1)
self.pwr = machine.Pin(P_Power_GSM,machine.Pin.OUT,machine.Pin.PULL_UP,hold=True)
# Let VBAT stabilize before PWRKEY - pulsing immediately
# after the rail comes up is ignored by the modem (cold
# first boot failed exactly this way in the field).
time.sleep(1)
self.wakeUp()
# Up to 3 PWRKEY attempts. The old retry loop here compared
# `at == None`, but us() never returns None (it returns the
# string "False") - so a missed first pulse was never retried.
attempt = 0
while attempt < 3:
self.key.value(1)
time.sleep(1)
self.key.value(0)
time.sleep(2)
at = self.us("AT",10,1,"CPIN: READY")
if self.isOn():
break
attempt += 1
print("Modem not responding - PWRKEY retry", attempt)
self.us("AT",2)
self.us('AT+QCCID')
#self.us("AT+CFUN=1",0.5)
#self.us('AT+CPSMS=0')
#self.us('AT+CEREG=0')
registrationStart = time.time()
return True
[docs]
def turnOff(self):
"""Shuts the modem down gracefully and cuts its power rail.
Sends ``AT+QPOWD`` first so the modem detaches cleanly, then
drives the power rail low and holds it through deep sleep.
Returns:
bool: True on success, False if the shutdown raised.
"""
try:
self.us("AT+QPOWD")
#self.key.value(1)
time.sleep(1)
#self.key.value(0)
self.pwr = machine.Pin(P_Power_GSM,machine.Pin.OUT,hold=False)
self.pwr.value(0)
self.pwr = machine.Pin(P_Power_GSM,machine.Pin.OUT,machine.Pin.PULL_UP,hold=True)
return True
except :
#self.pwr.value(0)
return False
[docs]
def reset(self):
"""Power-cycles the modem.
This board has no dedicated RESET_GSM pin, so instead of pulsing
a reset line the driver power-cycles ``P_Power_GSM``.
"""
self.turnOff()
self.pwr = machine.Pin(P_Power_GSM, machine.Pin.OUT, value=0, hold=True)
self.turnOn()
[docs]
def registerNetwork(self,registrationTimeout = 310):
"""Blocks until the modem registers on the network or times out.
Args:
registrationTimeout (int): Maximum seconds to wait, measured
from the global ``registrationStart`` stamped by
:meth:`turnOn`.
Returns:
bool: True when registered, False on timeout.
"""
global registrationTime
global registrationStart
if self.isReg():
#db.write('600',time.time() - registrationStart)
#self.us("AT+QEDRXCFG=0")
return True
else:
#First timer
while time.time() - registrationStart <registrationTimeout:
print("Reg timer 1 - remembered from previous session")
if self.isReg():
#db.write('600',time.time() - registrationStart)
registrationTime = time.time() - registrationStart
print("Registration Time ", registrationTime)
return True
else :
time.sleep(1)
return False
[docs]
def sleep(self):
"""Puts the modem into Quectel slow-clock mode.
Sends ``AT+QSCLK=1`` and raises DTR; the modem stays registered
but drops its supply current to the sleep floor.
"""
self.us('AT+QINDCFG="smsincoming",1', 0.5)
self.us("AT+QSCLK=1", 0.5)
self.dtr.value(1)
[docs]
def wakeUp(self):
"""Wakes the modem from slow-clock mode (DTR low + ``AT+QSCLK=0``)."""
self.dtr.value(0)
self.us('AT+QSCLK=0')
[docs]
def isReg(self):
"""Checks whether the modem is registered and has a usable IP.
Accepts CREG state 5 (roaming) or CEREG state 1 (home). State 3
logs a forbidden network, state 4 resets the modem, state 0
waits for the modem to start trying. When registered but the
PDP address is ``0.0.0.0``, the PDP context is bounced.
Returns:
bool: True when registered with a real IP address.
"""
global noReg
noReg = 0
# return True
if type(self.us("AT",0.1)) == bytes:
if '+CREG: 0,5' not in self.us('AT+CREG?') and '+CEREG: 0,1' not in self.us('AT+CREG?') :
noReg += 1
if '+CREG: 0,3' in self.us('AT+CREG?'):
self.us('AT+QEMMS?')
self.us('AT+QESMC?')
print("Forbidden network!")
#self.reset()
elif '+CREG: 0,4' in self.us('AT+CREG?'):
self.us('AT+QEMMS?')
self.us('AT+QESMC?')
print("Unknown status! Resetting modem...")
self.reset()
elif '+CREG: 0,0' in self.us('AT+CREG?'):
print("Modem is not trying to register!")
#self.us("AT+CFUN=1")
n=0
time.sleep(2)
while '+CREG: 0,0' in self.us('AT+CREG?') and n<50:
time.sleep(1)
n+=1
return False
else:
if "+CGPADDR: 1," in self.us("AT+CGPADDR"):
if '+CGPADDR: 1,"0.0.0.0"' in self.us("AT+CGPADDR"):
self.us("AT+CGACT=0,1")
time.sleep(1)
self.us("AT+CGACT=1,1")
return False
else:
noReg = 0
return True
else:
self.us("AT+CGACT=0,1")
time.sleep(1)
self.us("AT+CGACT=1,1")
return False
elif self.us("AT",0.1) == 'False' :
print("None from isReg()")
if self.isOn():
return False
else :
print("Restarting modem from isReg")
self.turnOff()
self.turnOn()
return False
[docs]
def getImei(self):
"""Reads the modem IMEI.
Also performs the one-time UART setup (PDP context for the APN,
echo, SMS text mode, URC routing to UART1) - which is why the
firmware calls it once during first boot.
Returns:
str: The IMEI (15 digits), or an empty string when the
modem is not responding (callers retry on ``len != 15``).
"""
self.us('AT+CGDCONT=1,"IP","'+apn+'"')
self.us("ATE1")
self.us('AT+CMGF=1', 0.5)
self.us('AT+QINDCFG="smsincoming",1', 0.5)
self.us('AT+QURCCFG="urcport","uart1"', 0.5)
#self.us("AT+QLEDMODE=1")
#self.us("AT+CFUN=0")
#time.sleep(5)
#AT+CPSMS
#self.us('AT+QCFG="EPCO",0')
#self.us("AT+CFUN=1")
#self.us('AT+QBAND=1,8')
#self.us('AT+QCGDEFCONT="IPV4V6","'+apn+'"')
#self.us('AT+QCFG="dsevent",0')
self.us("AT",2)
self.us("AT",2)
time.sleep(5)
self.us("AT",2)
raw = self.us("AT+CGSN=1")
if not hasattr(raw, "decode"):
# Modem silent: us() returned the "False" sentinel string.
# Return "" so initDefaults()'s len!=15 retry loop kicks in
# instead of crashing on str.decode.
return ""
imei = raw.decode("utf-8")
imei = imei.replace("AT+CGSN=1","")
imei = imei.replace("+CGSN: ","")
imei = imei.replace('"',"")
imei = imei.replace("\r","")
imei = imei.replace("\n","")
imei = imei.replace("OK","")
return imei
[docs]
def getBat(self):
"""Reads the battery voltage via ``AT+CBC``.
Returns:
list: Legacy shape ``[0, voltage_mV, '100']`` - index 1 is
the supply voltage in millivolts as a string.
"""
#return ['100','100']
self.us("AT")
bat = self.us('AT+CBC',1).decode("utf-8")
bat = bat.split('CBC: ')
#['AT+CBC\r\r\n+', '0,81,4008\r\n\r\nOK\r\n']
bat = bat[1].split(',')
bat = bat[2].split('\r\n')
#['0', '81', '4008\r\n\r\nOK\r\n']
#self.fillEng()
return [0,bat[0],'100']
[docs]
def getTimeF(self):
"""Fetches the current server time via the ``/t/`` endpoint.
Returns:
str: Epoch seconds as a 10-digit string, or False on error.
"""
self.us("AT")
self.connectHiGPS(1)
tim = self.sendHiGPS("/t/")
try:
return tim.decode("utf-8")
except:
return False
[docs]
def getCPSI(self):
"""Returns the raw engineering info (``AT+QENG=0``).
Returns:
str: The ``QENG`` payload with whitespace collapsed to
``;`` separators.
"""
self.us("AT")
eng = self.us("AT+QENG=0",1).decode("utf-8")
ee = eng.split("QENG:")
#self.us("AT+CENG=0")
eng = ee[1]
eng = eng.replace("\r","")
eng = eng.replace("\n","")
eng = eng.replace("+","")
eng = eng.replace(" ",";")
return eng
[docs]
def connectHiGPS(self,num):
"""Prepares the data connection to the HiGPS server.
Closes any stale socket, ensures network registration and
activates the PDP context.
Args:
num (int): Kept for compatibility, unused.
"""
connected = False
retry = 3
#self.us("AT+CGATT=1")
self.cipClose()
if self.registerNetwork():
self.us("AT+CGACT=1,1")
#cnact = self.us('AT+CGATT=1',3).decode("utf-8")
#while "CGATT: 1" in cnact and retry >0 :
# deact = self.us('AT+CGATT=0',2)
# act = self.us('AT+CGATT=1',3)
# cnact = self.us('AT+CGATT=1',3).decode("utf-8")
# retry = retry -1
#if "CGATT: 1" in cnact :
# self.turnOff()
# self.turnOn()
# self.registerNetwork()
[docs]
def getData(self,response):
#resp = ['AT+CARECV=0,100\r', '+CARECV: 7,*SET,55', '', 'OK', '']
#resp = ['AT+CARECV=0,100\r', '+CARECV: 0', '', 'OK', '', '+CADATAIND: 0', '', '+CASTATE: 0,0', '']
#resp = ['AT+CARECV=0,100\r', '+CARECV: 5,15858', '', 'OK', '']as
"""Extracts the payload from a raw ``+QIRD`` response.
Args:
response (bytes): Raw bytes as returned by ``AT+QIRD=0``.
Returns:
bytes: The payload; empty when nothing was received.
"""
splitString = b'+Q'
splitString += b'I'
splitString += b'R'
splitString += b'D'
splitString += b': '
try:
resp = response.split(splitString)
if len(resp) > 1:
dataLen =resp[1].split(b'\r\n')[0]
dataLengthInt = int(dataLen)
print("Data Length", dataLen)
startString = len(dataLen)+2
data = resp[1][startString:startString+dataLengthInt]
#dat =resp[1].split(b'\r\n')[1]
#print("dat",dat)
#data = dat[0:dataLengthInt ]
#print("Response Data", data)
else:
return ''
except:
return ''
print("Response Data", data)
return data
[docs]
def sendHiGPS(self,message,fast=False):
"""Sends a GET request to the HiGPS server and returns the reply.
Checks the socket with :meth:`socketState` and (re)opens it via
:meth:`openHiGPS` when it is not *Connected*, then sends
``GET <message>`` and extracts the payload from the ``+QIRD``
response.
Args:
message (str): Request path, e.g. ``'/t/'`` or
``'/input.php?IMEI=...'``. Truncated to 1400 chars.
fast (bool): Kept for compatibility, unused.
Returns:
bytes: The reply payload, or False on send failure.
"""
message = message.replace('"','')
print("modem.sendUDP message:" )
print(message)
tim = 0
if len(message) > 1400:
print("message_truncated")
message = message[0:1400]
passed = False
# 2 == Connected. Anything else (or None = no socket) -> (re)open.
# QIOPEN is asynchronous, so openHiGPS() polls until Connected and
# retries by itself; never fire QISEND into a socket that isn't up.
if self.socketState(0) != 2:
if not self.openHiGPS():
print("Could not open HiGPS socket")
return False
sendResult = self.us('AT+QISEND=0,'+str(len(message)+6),1)
if "ERROR" in sendResult:
return False
self.us('GET '+message+'\r\n')
#self.us('AT+CAACK=0',1)
#resp = self.us('AT+CARECV=0,512',1)
inData = b''
tim = 0
resp = b''
time.sleep(2)
inData = self.us('AT+QIRD=0')
print(inData)
#kew = "c"+"l"+'o'
#keyword = kew + "sed"
#while keyword not in resp and tim < 200 :
# tim = tim + 1
#print("Waiting for ", tim)
# inData += resp
# resp = self.uart.read()
# if resp is None:
# resp=b''
# time.sleep(0.1)
#QENG
#print("inData got in ", tim/10)
#print(inData)
try:
toret = self.getData(inData)
print("Toret = " ,toret)
return toret
except:
return False
[docs]
def socketState(self, connectID=0):
"""Returns the Quectel ``<socket_state>`` of a socket, or None.
Modem-agnostic replacement for substring-matching the raw
``AT+QISTATE`` reply. The ``+QISTATE:`` line is positional::
+QISTATE: <connectID>,<service_type>,<IP>,<remote_port>,
<local_port>,<socket_state>,<contextID>,<serverID>,
<access_mode>[,<AT_port>]
EG915 and EG916 differ only in the trailing fields and in the
echo/OK-only reply when no socket exists, so ``<socket_state>``
is read by field position (index 5) and the rest is ignored.
Args:
connectID (int): Socket index; 0 is the HiGPS connection.
Returns:
int: 0 Initial, 1 Opening, 2 Connected, 3 Listening,
4 Closing - or None when the socket does not exist or the
modem is silent.
"""
resp = self.us('AT+QISTATE=1,' + str(connectID), 1)
if not hasattr(resp, "decode"):
return None
for line in resp.decode("utf-8", "ignore").splitlines():
line = line.strip()
if not line.startswith("+QISTATE:"):
continue
fields = line.split(":", 1)[1].split(",")
if len(fields) < 6:
continue
try:
if int(fields[0]) != connectID:
continue
return int(fields[5])
except (ValueError, IndexError):
continue
return None
[docs]
def qiopenResult(self, resp):
"""Extracts ``<err>`` from a ``+QIOPEN: <connectID>,<err>`` reply.
Parses by field position, so it behaves identically on EG915 and
EG916 regardless of the surrounding echo/URC noise.
Args:
resp (bytes): Raw reply as returned by :meth:`us`.
Returns:
int: The result code (0 = success), or None when no
``+QIOPEN:`` line is present.
"""
if not hasattr(resp, "decode"):
return None
for line in resp.decode("utf-8", "ignore").splitlines():
line = line.strip()
if line.startswith("+QIOPEN:"):
parts = line.split(":", 1)[1].split(",")
try:
return int(parts[1])
except (ValueError, IndexError):
return None
return None
[docs]
def openHiGPS(self, retries=3, timeout=15):
"""Opens TCP socket 0 to ``application.higps.org:80``.
``AT+QIOPEN`` in buffer access mode is asynchronous: the modem
answers OK at once and emits ``+QIOPEN: 0,<err>`` seconds later,
when DNS and the TCP handshake finish. This waits for that URC
(up to ``timeout`` s), reads the real result code with
:meth:`qiopenResult` and only then confirms with
:meth:`socketState`. Every attempt starts from a closed socket,
because both modems reject QIOPEN on a connectID already in use.
Args:
retries (int): Open attempts before giving up.
timeout (int): Seconds to wait for the ``+QIOPEN:`` URC.
Returns:
bool: True once the socket is Connected (state 2).
"""
for attempt in range(retries):
# QIOPEN on a connectID that's already in use is rejected by both
# modems, so always start from a clean socket.
self.cipClose()
time.sleep(0.5)
resp = self.us('AT+QIOPEN=1,0,"TCP","application.higps.org",80',
timeout, 1, '+QIOPEN:')
err = self.qiopenResult(resp)
print("QIOPEN attempt", attempt, "result code:", err)
if err == 0 and self.socketState(0) == 2:
print("HiGPS socket connected")
return True
# err None -> modem stayed silent / URC never arrived (timeout)
# err != 0 -> open refused (e.g. 565/566 = DNS, 563 = PDP, ...)
print("QIOPEN failed - retrying open")
return False
[docs]
def cipClose(self):
# self.us("AT+QICLOSE=0")
"""Closes TCP socket 0 (``AT+QICLOSE=0``)."""
self.us("AT+QICLOSE=0")
#self.us("AT+CIPCLOSE")
[docs]
def qird(self, want, timeout=10):
"""Reads exactly ``<len>`` payload bytes with ``AT+QIRD=0,<want>``.
:meth:`us` is line-oriented and truncates a payload at the first
newline, which a binary firmware image contains - so OTA reads
must not go through it. This reads the ``+QIRD: <len>`` header
and then exactly ``<len>`` bytes off the UART.
Args:
want (int): Maximum bytes to request.
timeout (int): Seconds to wait for the data.
Returns:
bytes: The payload; empty when the socket has nothing yet
(``+QIRD: 0``) or on timeout.
"""
u = self.uart
u.read() # drop stale echo / URC noise
u.write('AT+QIRD=0,' + str(want))
u.write(bytes([0x0d, 0x0a]))
buf = b''
t = 0.0
while t < timeout:
r = u.read()
if r:
buf += r
i = buf.find(b'+QIRD:') # ':' -> response, not the '=' echo
if i >= 0:
j = buf.find(b'\r\n', i)
if j >= 0:
try:
n = int(buf[i + 6:j].strip())
except ValueError:
return b''
if n == 0:
return b''
payload = buf[j + 2:]
while len(payload) < n and t < timeout:
r2 = u.read()
if r2:
payload += r2
else:
utime.sleep(0.05)
t += 0.05
return payload[:n]
else:
utime.sleep(0.05)
t += 0.05
return b''
[docs]
def downloadChunk(self, path, expected, timeout=20):
"""Fetches one OTA chunk over the raw HiGPS socket.
Sends ``GET <path>`` and reads via :meth:`qird` in a loop until
``expected`` bytes arrive (LTE packets are far larger than the
single read that sufficed on 2G) or the socket idles past
``timeout``.
Args:
path (str): Repository path, e.g. ``/repo/<name>/1460/<n>/<imei>``.
expected (int): Chunk size the caller expects.
timeout (int): Idle seconds before giving up.
Returns:
bytes: Whatever arrived. A caller that gets fewer than
``expected`` bytes must retry the same chunk index and never
advance the file, so a short read cannot misalign the
download.
"""
if self.socketState(0) != 2:
if not self.openHiGPS():
return b''
msg = path.replace('"', '')
send = self.us('AT+QISEND=0,' + str(len(msg) + 6), 1)
if "ERROR" in send:
return b''
self.us('GET ' + msg + '\r\n')
data = b''
idle = 0.0
while len(data) < expected and idle < timeout:
part = self.qird(expected - len(data))
if part:
data += part
idle = 0.0
else:
time.sleep(0.5)
idle += 0.5
return data
[docs]
def getVersion(self):
"""Returns the modem firmware version string (``AT+QGMR``).
Returns:
str: Version line, or an empty string when unavailable.
"""
version = self.us('AT+QGMR',1)
lines = version.decode().splitlines()
if len(lines) < 3:
return ""
return lines[2]
[docs]
def getNetworkInfo(self):
"""Reads the serving-cell information (LTE only).
Parses ``AT+QENG="servingcell"``.
Returns:
dict: Keys ``NetworkBearer``, ``Band``, ``SignalRSSI``,
``RSRP``, ``RSRQ``, ``SINR``, ``LinkQuality``, ``CellID``,
``SMNC``, ``SMCC``, ``LAC``, ``PCI``, ``EARFCN``,
``TimingAdvance``. False when not on an LTE cell.
"""
info = {
"NetworkBearer": "None",
"Band": "None",
"SignalRSSI": "None",
"RSRP": "None",
"RSRQ": "None",
"SINR": "None",
"LinkQuality": "None",
"CellID": "None",
"SMNC": "None",
"SMCC": "None",
"LAC": "None",
"PCI": "None",
"EARFCN": "None",
"TimingAdvance": "None"
}
resp = self.us('AT+QENG="servingcell"', 0.5)
if resp and isinstance(resp, bytes):
try:
lines = resp.decode().splitlines()
for line in lines:
if "+QENG:" in line and "LTE" in line:
parts = line.split(",")
if len(parts) >= 18:
info["NetworkBearer"] = parts[2].strip().strip('"') # LTE
info["Band"] = parts[9].strip()
info["SMCC"] = parts[4]
info["SMNC"] = parts[5]
info["CellID"] = parts[6]
info["PCI"] = parts[7]
info["EARFCN"] = parts[8]
info["LAC"] = parts[12]
info["RSRP"] = parts[13]
info["RSRQ"] = parts[14]
info["SignalRSSI"] = parts[15]
info["SINR"] = parts[16]
info["TimingAdvance"] = parts[17]
info["LinkQuality"] = parts[13]
except Exception as e:
print("Parse error in getNetworkInfo():", e)
if info["NetworkBearer"] is "None":
return False
else:
return info
[docs]
def turnOnGPS(self):
"""Enables the built-in GNSS receiver (``AT+QGPS=1``)."""
self.us("AT+QGPS=1")
[docs]
def turnOffGPS(self):
"""Disables the built-in GNSS receiver (``AT+QGPS=0``)."""
self.us("AT+QGPS=0")
[docs]
def gpsIsFixed(self):
"""Checks whether the GNSS receiver has a position fix.
Returns:
bool: False while the modem answers ``+CME ERROR: 516``.
"""
if "+CME ERROR: 516" in self.us("AT+QGPSLOC=2"):
return False
return True
[docs]
def getGPS(self):
#+QGPSLOC: 194918.00 ,42.66993 ,23.23396 ,1.14, 739.7 , 3 , ,0.724, 0.391 ,260725, 16
#+QGPSLOC: <UTC> ,<latitude>,<longitude>,<HDOP>,<altitude>,<fix>,<COG>,<spkm>,<spkn>,<date>,<nsat>
# 0 1 2 3 4 5 6 7 8 9 10
"""Reads the current GNSS position (``AT+QGPSLOC=2``).
Returns:
dict: Keys ``time``, ``latitude``, ``longitude``, ``HDOP``,
``altitude``, ``fix``, ``COG``, ``spkm``, ``spkn``,
``date``, ``nsat``, ``errcode``. False when there is no fix.
"""
gpsData = self.us("AT+QGPSLOC=2")
if "+CME ERROR: 516" in gpsData:
return False
info = {
"time" : None,
"latitude": None,
"longitude": None,
"HDOP": None,
"altitude": None,
"fix": None,
"COG": None,
"spkm": None,
"spkn": None,
"date": None,
"nsat": None,
"errcode": None,
}
if gpsData and isinstance(gpsData, bytes):
try:
lines = gpsData.decode().splitlines()
for line in lines:
line = line.strip(' ')
parts = line.split(",")
if len(parts) >= 10:
info["time"] = parts[0].split(":")[1].strip(' ') #+QGPSLOC: 194626.00
info["latitude"] = parts[1]
info["longitude"] = parts[2]
info["HDOP"] = parts[3]
info["altitude"] = parts[4]
info["fix"] = parts[5]
info["COG"] = parts[6]
info["spkm"] = parts[7]
info["spkn"] = parts[8]
info["date"] = parts[9]
info["nsat"] = parts[10]
except Exception as e:
print("Parse error in getGPS():", e)
return info
[docs]
def check_inbox(self):
"""Lists the SMS messages stored on the modem.
Returns:
list: One dict per message with keys ``index``, ``sender``,
``date`` and ``message``.
"""
self.us('AT+CMGF=1', 0.5)
self.us('AT+CPMS="ME","ME","ME"', 0.5)
messages = []
resp = self.us('AT+CMGL="ALL"', 1)
if not resp or not isinstance(resp, bytes):
return messages
lines = resp.decode().splitlines()
i = 0
while i < len(lines):
line = lines[i].strip()
if line.startswith("+CMGL:"):
try:
# Example: +CMGL: 1,"REC UNREAD","+359888123456","","2025/07/26,18:30:55+08"
parts = line.split(",")
idx = int(parts[0].split(":")[1])
sender = parts[2].replace('"', '')
date = parts[4].replace('"', '') + " " + parts[5].replace('"', '')
if i + 1 < len(lines):
msg = lines[i + 1].strip()
else:
msg = ""
messages.append({
"index": idx,
"sender": sender,
"date": date,
"message": msg
})
except Exception as e:
print("Error parsing SMS:", e)
i += 1
return messages
[docs]
def delete_sms(self, index):
"""Deletes an SMS from the modem memory.
Args:
index (int): Message index as reported by :meth:`check_inbox`.
"""
self.us(f'AT+CMGD={index}', 0.5)
[docs]
def send_sms(self, phone, text):
"""Sends an SMS.
Args:
phone (str): Destination number in international format.
text (str): Message body.
Returns:
bool: Always True (the modem confirms asynchronously).
"""
self.us('AT+CMGF=1', 0.5)
self.us(f'AT+CMGS="{phone}"', 0.5)
self.uart.write(text.encode('utf-8') + b"\x1A") # Ctrl+Z to end the SMS
time.sleep(3)
return True
[docs]
def us(self,arg,t=0,t2=0,expect=None):
#print("Expecting ",expect)
#print(arg)
"""Sends a raw AT command and collects the reply.
Args:
arg (str or bytes): The command, without the trailing CRLF.
t (float): Maximum extra seconds to poll for a reply.
t2 (float): Fixed sleep before the first read.
expect (str or bytes): Optional substring; polling stops
early once it appears in the reply.
Returns:
bytes: The raw reply, or the string ``"False"`` when
nothing arrived (legacy sentinel).
"""
answer = []
self.uart.write(arg)
self.uart.write(bytes([0x0d,0x0a]))
utime.sleep(0.1)
tstart = 0
if t2 >0:
utime.sleep(t2)
answer = self.uart.read()
while tstart <= t:
#print("Answer debug ", answer)
if answer == None:
utime.sleep(0.5)
answer = self.uart.read()
tstart += 0.5
elif b'\n' not in answer:
utime.sleep(0.1)
newInput = self.uart.read()
if newInput != None:
answer +=newInput
tstart += 0.1
else:
if expect != None:
if expect in answer:
#print("Found expected answer ", answer)
break
else:
utime.sleep(0.1)
#print("Answer debug ", answer)
answer = self.uart.read()
tstart += 0.1
else:
break
print(answer)
if hasattr(answer,"decode"):
return answer
else:
return "False"
[docs]
def us2(self,arg,t=0):
#print(arg)
"""Simplified one-shot variant of :meth:`us`.
Args:
arg (str or bytes): The command, without the trailing CRLF.
t (float): Fixed seconds to wait before reading the reply.
Returns:
bytes: The raw reply, or ``"False"`` when nothing arrived.
"""
answer = []
self.uart.write(arg)
self.uart.write(bytes([0x0d,0x0a]))
time.sleep(0.1)
time.sleep(t)
answer = self.uart.read();
if answer == None:
time.sleep(0.5)
answer = self.uart.read();
print(answer)
if hasattr(answer,"decode"):
return answer
else:
return "False"