Method To Take Control of Cell Modem
-
Improved http_post() function
This version can receive large files. Tested with file size > 128,000 bytes.
##################################################################### # HTTP Post ##################################################################### def http_post(uri, header, tx_data): # --> AT+SQNHTTPSND=<prof_id>,<command>,<resource>,<data_len>[,<post_param>[,<extra_header_line>]] # xx> POST <uri> HTTP/1.1 # xx> <header> # xx> Content-Length: <tx_size> # xx> Content-Type: application/octet-stream # xx> <blank line> # <-- '> ' # --> <tx_data bytes> # <-- OK # <-- +SQNHTTPRING: <prof_id,<http_status_code>,<content_type>,<rx_size> # --> AT+SQNHTTPRCV=1,0\r # <-- /r/n<<< # <-- <rx_data bytes> # <-- OK try: log("HTTP post") while True: # note: init_modem() sets configuration "AT+SQNHTTPCFG=..." at startup tx_size = len(tx_data) # log("writing message: {0}".format(tx_data)) log("tx_data length: {0}".format(tx_size)) # AT+SQNHTTPSND=<prof_id>,<command>,<resource>,<data_len>[,<post_param>[,<extra_header_line>]] # prof_id=1, command=0 (POST), <uri>, <tx_size>, post_parm="2" (application/octet-stream), header="Host: <host_name>" a = at('AT+SQNHTTPSND=1,0,"{0}",{1},"2","{2}"'.format(uri, tx_size, header)) if a[0] != '> ': break # error uart.write(tx_data) a = get_modem_response() if a[0] != 'OK': break # error # modem responds with "+SQNHTTPRING: <prof_id,<http_status_code>,<content_type>,<data_size>" a = get_modem_response() s = a[0] if s[0:13] != "+SQNHTTPRING:": break # error a = s.split(",") http_status_code = int(a[1]) if http_status_code != 200: break # error rx_size = int(a[3]) log("rx_data length: {0}".format(rx_size)) # get response cmd = 'AT+SQNHTTPRCV=1,0\r' log(" modem command: {0}".format(cmd)) uart.write(cmd) r = wait_rx(5, 10000) # wait for b'\r\n<<<' if r == False: break # error b = uart.read(5) # log(" modem response: {0}".format(b)) if b[2:5] != b'<<<': break # error t0 = ticks_ms() b = b'' n = rx_size # no. bytes remaining to be read while True: if uart.any() > 0: log(" attempt to read {0} bytes".format(n)) bx = uart.read(n) log(" bytes read: {0}".format(len(bx))) # log(" modem response: {0}".format(bx)) b = b + bx if len(b) == rx_size: break # entire message received n = rx_size - len(b) t0 = ticks_ms() # reset timeout t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > 10000: log("timeout error receiving response") return b'' sleep_ms(400) log(" total bytes read: {0}".format(len(b))) # log(" post response: {0}".format(b)) a = get_modem_response() if a[0] != 'OK': break # error # log(b) return b except Exception as ex: sys.print_exception(ex) # error return b'' -
@tlanier said in Method To Take Control of Cell Modem:
I need to understand what AT commands are needed to properly initialize the modem and set the APN based on the SIM card. This is an area I'm currently fuzzy about. Do bands need to be set? This is stuff that is handled during the creation of the LTE module which I no longer want to use. The answers may come from studying the C code of the LTE module. Currently the code just sets CFUN=1 and everything magically works.
I use this for the first time initializing:
send_cmd('AT+CGDCONT=1,"IP","cdp.iot.t-mobile.nl"') # set context 1 to ip and apn to cdp.iot.t-mobile.nl send_cmd("AT+SQNAUTOINTERNET=1") # maybe not needed for you use case? send_cmd('AT!="clearscanconfig"') # clear band scan config send_cmd('AT!="RRC:addScanBand band=8"') # scan for band 8 send_cmd('AT+COPS=1,2,"20416"') # manually register to plmn 20416 (tmobile nl) #send_cmd('AT+COPS=0') # automatically choose plmn -
@iot2020, thanks for the suggestions. It's interesting that when I search the Monarch-LR5110 manual neither of the AT!= commands are found.
-
@tlanier said in Method To Take Control of Cell Modem:
SQNHTTPSND
Hey @tlanier Hope you are doing well. Regarding "Tested with file size > 128,000 bytes" Just curious, how it is able to send such large payload at one go? Is it using fragmentation ? and divide the payload into multiple packets? and then send?
-
@bitvijays HTTP is based on TCP, so any HTTP payload would indeed be sent as multiple packets (the size and number of which depend on the path MTU). TCP normally makes use of windowing, so it would not even necessarily send all at once, it would wait for ACKs along the way and send more as the first ACKs are received.
It's interesting how this is all built into the modem itself...
-
@bitvijays In my case the response to a HTTP Post is a large file (208K file used to update the program). The program hangs in a loop retrieving whatever is in the UART receive buffer until the correct number of bytes is received (or timeout occurs).
-
@jcaron Agreed. Also, just wanted to confirm about the MTU. Would the MTU for NB-IoT be 680 bits (85 bytes) DL and 1000 bits (125 bytes) UL. Referred Section 8.2 NB-IoT Deployment Guide
-
A new method has been added to the modem module. It is desirable to use the ICCID to determine which APN to set. One issue to deal with is apparently the ICCID is not valid until registration with the tower is done. The set_apn() method should be called right before doing the attach().
To reduce the wear of writing to the non-volatile memory, I read the current value of the APN and skip the write if not necessary. I wonder if the modem does this type of checking internally? Ideally we could just set the APN once at program start, but we can't read the ICCID at that point for some reason. Even though the manual says when CFUN=4 or CFUN=1 access to the SIM is available, reading the ICCID does not work until after registration.
To get some context as to how the modem module can be used, I've included example code for a modem_ready() method that uses the modem module.
There is still work to be done with the basic at() and get_modem_response() methods. Unsolicited responses are causing problems in some cases. That will be my next area to concentrate on. There seems to be two paths to take when talking to a modem. Path #1 sends a command to the modem and then waits a period of time and assumes the modem has had time to respond completely. Path #2 sends a command to the modem and then watches the data being returned looking for (1) OK, or (2) ERROR anywhere in the response. The responses then must be filtered to only return the particular response line that is desired.
##################################################################### # Set APN ##################################################################### def set_apn(): try: log("set APN") # ICCID is only valid after registration r = iccid()[0:5] if r == "89014": apn = g.oIni.apn_89014 # Kore AT&T SIM elif r == "89012": apn = g.oIni.apn_89012 # Kore T-Mobile SIM elif r == "89445": apn = g.oIni.apn_89445 # Hologram SIM else: apn = g.oIni.apn_89014 # Kore AT&T SIM g.oIni.apn = apn log("APN: {0}".format(apn)) # try to reduce wear of non-volatile memory a = at('AT+CGDCONT?') # response --> ['+CGDCONT: 1,"IP","10569.mcs",,,,0,0,0,0,0,0', ...] s = a[0] a = s.split(",") if (a[0] == '+CGDCONT: 1') and (a[1] == '"IP"') and (a[2] == '"'+apn+'"'): return # apn needs setting a = at('AT+CGDCONT=1,"IP","{0}"'.format(apn)) if a[0] != 'OK': log("error setting APN") except Exception as ex: sys.print_exception(ex) ##################################################################### # Modem Ready ##################################################################### def modem_ready(): try: log("** check for modem ready") # get modem signal strength g.oStatus.cell_signal = modem.signal_strength() # handle SMS messages r = modem.sms_read() if len(r) > 1: modem.sms_process(r) modem.sms_delete() # check modem registration r = modem.check_registration() if (r == 1) or (r == 5): # wait until registered to set APN modem.set_apn() modem.attach() if modem.isattached(): g.com_secs = 0 return True reset = False if g.com_secs > (10*60): # unable to attach in 10 mins, try resetting modem reset = True if modem.restarted(): # modem has crashed / restarted, try resetting modem reset = True if reset: modem.reset_modem() modem.init_modem() g.com_secs = 0 return False except Exception as ex: sys.print_exception(ex) return False -
@tlanier you should probably create a GitHub or GitLab repo for your code...
-
@jcaron Thanks for the suggestion. I have been thinking about that. I would need to make the module more general purpose and less specific to my application. My first priority is to make our application more robust and error free.
I've thought about creating a general purpose do nothing application that implements multi-threading, a timer interrupt service routine, and the modem module to show how everything fits together. Someone could take the template application and create a full application in no time.
Including our method of program updates would also be useful I think.
-
The new at(cmd, filter="") function is now improved such that unsolicited responses should have no affect. It also allows you to specify the response you are interested in getting back by setting the filter parameter. This simplifies the coding of the rest of the modem module.
##################################################################### # Send AT Command To Modem ##################################################################### def at(cmd, filterx=""): try: if uart.any() > 0: # get unsolicited modem response a = at_response() for s in a: at_process(s) log(" modem command: {}".format(cmd)) # send command to modem n = uart.write(cmd + '\r') if n != len(cmd) + 1: log("error sending command to modem: n = {}".format(n)) return [] a = at_response(filterx) return a except Exception as ex: sys.print_exception(ex) a = [] return a # return modem response lines in array ##################################################################### # Filter Modem Response ##################################################################### def at_filter(a, filterx): # a is passed as reference value if filterx == "": return # filter response n = len(filterx) i = len(a) while i > 0: i = i - 1 if a[i][0:n] != filterx: # remove lines that do not start with filter value at_process(a[i]) a.pop(i) ##################################################################### # Process Unsolicited Response ##################################################################### def at_process(s): if s == 'OK': pass else: log("unsolicited modem response: {}".format(s)) return ##################################################################### # Get AT Command Response ##################################################################### def at_response(filterx=""): try: t0 = ticks_ms() while True: if uart.any() > 0: # insure entire response has been received sleep_ms(50) break t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > 10000: log("timeout error - no modem response after 10 secs") return [] sleep_ms(100) b = uart.read() s = str(b, 'utf-8') a = s.split('\r\n') a = list(filter(None, a)) log(" modem response: {}".format(a)) at_filter(a, filterx) except Exception as ex: sys.print_exception(ex) a = [] return a # return modem response lines in array -
Hey @tlanier could you elaborate a bit on this point? How to reproduce this freezing?
Task switching no longer freezes during modem communication.
-
Here's some of the higher level code that uses the modem module.
import globals as g import modem from poll import poll import pycom import sys import uos import util from util import log from utime import sleep_ms import wifi ##################################################################### # Communication Task ##################################################################### def com_task(): try: sleep_ms(500) log("com_task: started") wifi.deinit() modem.reset_modem() modem.init_modem() g.poll_host = True # triggers poll at startup except Exception as ex: sys.print_exception(ex) while True: try: if modem_ready(): if ready_to_poll(): modem.set_apn() transmit() sleep_ms(2000) except Exception as ex: sys.print_exception(ex) ##################################################################### # Modem Ready ##################################################################### def modem_ready(): try: log("** check for modem ready") # get modem signal strength g.oStatus.cell_signal = modem.signal_strength() # check modem registration r = modem.check_registration() if (r == 1) or (r == 5): # registered g.com_secs = 0 # handle SMS messages r = modem.sms_read() if len(r) > 1: modem.sms_process(r) modem.sms_delete() return True reset = False if g.com_secs > (10*60): # unable to attach in 10 mins, try resetting modem reset = True if modem.restarted(): # modem has crashed / restarted, try resetting modem reset = True if reset: modem.reset_modem() modem.init_modem() g.com_secs = 0 return False except Exception as ex: sys.print_exception(ex) return False -
@peterp Sorry to take so long to respond. I wanted to finish updating my posts showing the new code that seems to be working well. Soon I should be able to post the entire refined code. Maybe it will help others with the same problems we have had.
The reason I went down this development path was as described originally:
"I currently have a program working in production using Pycom MicroPython V1.20.0.rc13 on many units. When I try to run the same program on a newly purchased GPy running Pycom MicroPython 1.20.2.r3, the program crashes (20 sec watchdog timeout) when I do an AT command ('AT+CEREG?') after disconnecting from a successful transmission. How do I troubleshoot such a crash?"
Our original reason for picking Pycom and GPy was the thinking that all the modem complexity would be handled by the high level LTE module. It's only a guess but our thinking that since the crashes always happened during times communicating with the modem, the LTE module was the likely source of the problem.
The new method seems to be superior from the standpoint of interference with multiple tasks. Since we always call non-blocking uart commands, the code should never get hung in the C code for any period. A simple example of the LTE module code freezing multi-tasking is to do a simple modem reset. Reading and writing to the file storage also has a minor affect. The real reason for the development was to try to eliminate program crashes. Only longer term testing will show whether we have achieved that goal.
One concern I do have is whether I am properly initializing the cell modem. Any input from anybody out there that is more familiar with properly setting up the LTE modem would be appreciated. It seems to work fine at this point; however, I'm not sure whether it's working because the GPy I'm testing with has run the LTE initialization at some point in the past and that has possibly setup the non-volatile memory of the modem to the proper settings that are needed.
One drawback of using this method of transmission is it is not compatible with WIFI connections. That is not an issue for us at this time. Maybe later we can research how to do some form of communication that is compatible with both WIFI and direct AT commands. I assume that involves using the socket commands of the modem, but this is an area I'm not sure about at this time.
-
@tlanier Admire your efforts squire. I've been struggling with lte for a year or so now. It's not perfect (the lte.isattached() gets confused sometimes) but I got it robust enough to be usable. I use the requests library for data transfer & my main complaint is that it can't be timed out. I modified it with machine.WDT so that I can abort in a timely manner but I'd like a better solution. Am I right in thinking your wait_rx(n, 10000) allows you to quit if http post/get times out after 10s?
-
@kjm I think I remember having to modify urequests also. Here's what I currently use in my old version that is working.
import usocket import util class Response: def __init__(self, f): self.raw = f self.encoding = "utf-8" self._cached = None def close(self): if self.raw: self.raw.close() self.raw = None self._cached = None @property def content(self): if self._cached is None: try: self._cached = self.raw.read() finally: self.raw.close() self.raw = None return self._cached @property def text(self): return str(self.content, self.encoding) def json(self): import ujson return ujson.loads(self.content) def request(method, url, data=None, json=None, headers={}, stream=None): try: proto, dummy, host, path = url.split("/", 3) except ValueError: proto, dummy, host = url.split("/", 2) path = "" if proto == "http:": port = 80 elif proto == "https:": import ussl port = 443 else: raise ValueError("Unsupported protocol: " + proto) if ":" in host: host, port = host.split(":", 1) port = int(port) # ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM) # PATCH ai = usocket.getaddrinfo(host, port) ai = ai[0] s = usocket.socket(ai[0], ai[1], ai[2]) try: s.settimeout(20) s.connect(ai[-1]) if proto == "https:": s = ussl.wrap_socket(s, server_hostname=host) s.write(b"%s /%s HTTP/1.0\r\n" % (method, path)) if not "Host" in headers: s.write(b"Host: %s\r\n" % host) # Iterate over keys to avoid tuple alloc for k in headers: s.write(k) s.write(b": ") s.write(headers[k]) s.write(b"\r\n") if json is not None: assert data is None import ujson data = ujson.dumps(json) s.write(b"Content-Type: application/json\r\n") if data: s.write(b"Content-Length: %d\r\n" % len(data)) s.write(b"\r\n") if data: s.write(data) l = s.readline() #print(l) l = l.split(None, 2) status = int(l[1]) reason = "" if len(l) > 2: reason = l[2].rstrip() while True: l = s.readline() if not l or l == b"\r\n": break #print(l) if l.startswith(b"Transfer-Encoding:"): if b"chunked" in l: raise ValueError("Unsupported " + l) elif l.startswith(b"Location:") and not 200 <= status <= 299: raise NotImplementedError("Redirects not yet supported") # except OSError: # PATCH except: s.close() raise resp = Response(s) resp.status_code = status resp.reason = reason return resp def head(url, **kw): return request("HEAD", url, **kw) def get(url, **kw): return request("GET", url, **kw) def post(url, **kw): return request("POST", url, **kw) def put(url, **kw): return request("PUT", url, **kw) def patch(url, **kw): return request("PATCH", url, **kw) def delete(url, **kw): return request("DELETE", url, **kw) -
@kjm Thanks for the comment. The intention of wait_rx(n, timeout=10000) is to be a non-blocking call that waits for N chars to come into the uart or to timeout after 10,000 msecs by default. N should be less than the buffer size of the uart which is currently set to 512 bytes.
In looking at my current code, I see a bug. I should have had a sleep command in the while loop to make multi-tasking more efficient. Here's the corrected code. Thanks again!
##################################################################### # Wait For N Received Chars ##################################################################### def wait_rx(n, timeout=10000): # timeout in msecs t0 = ticks_ms() while True: if uart.any() >= n: break t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > timeout: return False sleep_ms(10) return True -
@kjm What would really be neat is if we could learn how to put the modem in a mode that allowed the urequests/usockets libraries to work using just AT commands!
I suspect it has to do with the AT+CGDCONT parameter named PDP_type. The manual shows the following values:
X.25 ITU-T/CCITT X.25 layer 3 (Obsolete)
IP Internet Protocol (IETF STD 5 [103])
IPV6 Internet Protocol, version 6 (see RFC 2460 [106])
IPV4V6 Virtual <PDP_type> introduced to handle dual IP stack UE capability. (See
3GPP TS 24.301 [83])
OSPIH Internet Hosted Octect Stream Protocol (Obsolete)
PPP Point to Point Protocol (IETF STD 51 [104])
Non-IP Transfer of Non-IP data to external packet data network (see 3GPP TS 23.401This is an area I'm unsure about. I would welcome comments from others about what would need to be done.
-
Here's updated code that is now running on two units overnight with good success. The code handles unsolicited responses gracefully. The code is non-intrusive to other multi-tasking threads. The code has not crashed (more testing to be done for sure). The code is easy to understand; everything is there to see and improve (no hidden commands being sent to the modem).
# MODEM.PY Module import globals as g from machine import UART import micropython import sys import util from util import log from utime import sleep_ms, ticks_ms, ticks_diff uart = None ##################################################################### # Send AT Command To Modem ##################################################################### def at(cmd, filterx=""): try: if uart.any() > 0: # get unsolicited modem response a = at_response() for s in a: at_process(s) log(" modem command: {}".format(cmd)) # send command to modem n = uart.write(cmd + '\r') if n != len(cmd) + 1: log("error sending command to modem: n = {}".format(n)) return [] a = at_response(filterx) return a except Exception as ex: sys.print_exception(ex) a = [] return a # return modem response lines in array ##################################################################### # Filter Modem Response ##################################################################### def at_filter(a, filterx): # a is passed as reference value if filterx == "": return # filter response n = len(filterx) i = len(a) while i > 0: i = i - 1 if a[i][0:n] != filterx: # remove lines that do not start with filter value at_process(a[i]) a.pop(i) ##################################################################### # Process Unsolicited Response ##################################################################### def at_process(s): if s == 'OK': pass else: log("unsolicited modem response: {}".format(s)) return ##################################################################### # Get AT Command Response ##################################################################### def at_response(filterx=""): try: t0 = ticks_ms() while True: if uart.any() > 0: # insure entire response has been received sleep_ms(50) break t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > 10000: log("timeout error - no modem response after 10 secs") return [] sleep_ms(100) b = uart.read() s = str(b, 'utf-8') a = s.split('\r\n') a = list(filter(None, a)) log(" modem response: {}".format(a)) at_filter(a, filterx) except Exception as ex: sys.print_exception(ex) a = [] return a # return modem response lines in array ##################################################################### # Attach Cellular Modem To Packet Domain Service ##################################################################### def attach(): try: log("attach modem to packet domain service") a = at('AT+CGATT=1','OK') if len(a) == 0: log("error attaching modem") except Exception as ex: sys.print_exception(ex) ##################################################################### # Check Registration ##################################################################### def check_registration(): # return 0-5, 99-unknown # response: ['+CEREG: 2,4', 'OK'] # response: ['+CEREG: 2,1,"2C29","0B54400F",7', 'OK'] # 2,0 - not registered, not searching # 2,1 - registered, home network # 2,2 - not registered, searching # 2,3 - registration denied # 2,4 - unknown (out of coverage) # 2,5 - registered, roaming try: while True: log("check registration") a = at('AT+CEREG?','+CEREG:') if len(a) == 0: break a = a[0].split(",") if (a[0] != "+CEREG: 2"): break reg = int(a[1]) if reg == 1: log("registered - home network") elif reg == 5: log("registered - roaming") else: log("not registered") return reg except Exception as ex: log("check registration exception") sys.print_exception(ex) # unknown registration log("error checking registration") return 99 ##################################################################### # Detach Cellular Modem From Packet Domain Service ##################################################################### def detach(): # modem does not get SMS messages while the modem is detached # old messages received while detached are only received when a new message is sent while attached try: log("detach modem from packet domain service") a = at('AT+CGATT=0','OK') if len(a) == 0: log("error detaching modem") except Exception as ex: sys.print_exception(ex) ##################################################################### # Disable LTE Modem ##################################################################### def disable_modem(): try: log("disable modem") a = at('AT+CFUN=0','OK') # minimum functionality if len(a) == 0: log("error disabling modem") except Exception as ex: sys.print_exception(ex) ##################################################################### # Enable Modem ##################################################################### def enable_modem(): try: log("enabling modem") a = at('AT+CFUN=1','OK') # full functionality if len(a) == 0: log("error enabling modem") except Exception as ex: sys.print_exception(ex) ##################################################################### # Get CFUN - Modem Level of Functionality ##################################################################### def get_cfun(): # 0 - minimum functionality # 1 - full functionality # 4 - radios disabled but access to SIM allowed # -1 - unknown # note: if modem does +SHUTDOWN/+SYSSTART, CFUN will be reset to 0 try: while True: a = at('AT+CFUN?','+CFUN:') # response: ['+CFUN: 1', 'OK'] if len(a) == 0: break cfun = int(a[0][7]) return cfun except Exception as ex: sys.print_exception(ex) return -1 ##################################################################### # Get SMS Mode ##################################################################### def get_sms_mode(): # return 0-PDU mode or unknown, 1-text mode try: log("get SMS mode") a = at('AT+CMGF?','+CMGF:') # response: ['+CMGF: 1', 'OK'] if len(a) == 0: return 0 # mode unknown mode = int(a[0][7]) return mode except Exception as ex: sys.print_exception(ex) return 0 # mode unknown ##################################################################### # HTTP Post ##################################################################### def http_post(uri, header, tx_data): # --> AT+SQNHTTPSND=<prof_id>,<command>,<resource>,<data_len>[,<post_param>[,<extra_header_line>]] # xx> POST <uri> HTTP/1.1 # xx> <header> # xx> Content-Length: <tx_size> # xx> Content-Type: application/octet-stream # xx> <blank line> # <-- '> ' # --> <tx_data bytes> # <-- OK # <-- +SQNHTTPRING: <prof_id,<http_status_code>,<content_type>,<rx_size> # --> AT+SQNHTTPRCV=1,0\r # <-- /r/n<<< # <-- <rx_data bytes> # <-- OK try: log("HTTP post") while True: # note: init_modem() sets configuration "AT+SQNHTTPCFG=..." at startup tx_size = len(tx_data) # log("writing message: {0}".format(tx_data)) log("tx_data length: {0}".format(tx_size)) # AT+SQNHTTPSND=<prof_id>,<command>,<resource>,<data_len>[,<post_param>[,<extra_header_line>]] # prof_id=1, command=0 (POST), <uri>, <tx_size>, post_parm="2" (application/octet-stream), header="Host: <host_name>" a = at('AT+SQNHTTPSND=1,0,"{0}",{1},"2","{2}"'.format(uri, tx_size, header),'> ') if len(a) == 0: break # error uart.write(tx_data) a = at_response('OK') if len(a) == 0: break # error # modem responds with "+SQNHTTPRING: <prof_id,<http_status_code>,<content_type>,<data_size>" a = at_response("+SQNHTTPRING:") if len(a) == 0: break # error a = a[0].split(",") http_status_code = int(a[1]) if http_status_code != 200: break # error rx_size = int(a[3]) log("rx_data length: {0}".format(rx_size)) # get response cmd = 'AT+SQNHTTPRCV=1,0\r' log(" modem command: {0}".format(cmd)) uart.write(cmd) r = wait_rx(5, 10000) # wait for b'\r\n<<<' if r == False: break # error b = uart.read(5) # log(" modem response: {0}".format(b)) if b[2:5] != b'<<<': break # error t0 = ticks_ms() b = b'' n = rx_size # no. bytes remaining to be read while True: if uart.any() > 0: log("attempt to read {0} bytes".format(n)) bx = uart.read(n) log("bytes read: {0}".format(len(bx))) # log(" modem response: {0}".format(bx)) b = b + bx if len(b) == rx_size: break # entire message received n = rx_size - len(b) t0 = ticks_ms() # reset timeout t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > 10000: log("timeout error receiving response") return b'' sleep_ms(400) log("total bytes read: {0}".format(len(b))) # log(" post response: {0}".format(b)) a = at_response('OK') if len(a) == 0: break # error # log(b) return b except Exception as ex: sys.print_exception(ex) # error return b'' ##################################################################### # Get SIM Card ICCID Number ##################################################################### def iccid(): # CFUN must be 1 or 4 try: log("get SIM card ICCID") a = at('AT+SQNCCID?','+SQNCCID:') # example response: ['+SQNCCID: "89014103271203065543",""', 'OK'] if len(a) == 0: return '' a = a[0].split(",") n = len(a[0]) r = a[0][11:n-1] except Exception as ex: sys.print_exception(ex) r = '' log("ICCID = {0}".format(r)) return r # ICCID as string ##################################################################### # Get Modem IMEI Number ##################################################################### def imei(): try: log("get modem IMEI number") a = at('AT+CGSN=1','+CGSN:') # example response: ['+CGSN: "354347094028575"', 'OK'] if len(a) == 0: return '' n = len(a[0]) r = a[0][8:n-1] except Exception as ex: sys.print_exception(ex) r = '' log("IMEI = {0}".format(r)) return r # IMEI as string ##################################################################### # Initialize Modem ##################################################################### def init_modem(): try: log("initializing modem") if uart == None: init_uart() disable_modem() # set profile #1 parameters for HTTP connection # prof_id=1 (profile), ip, port, auth_type=0 (None), username="", password="", # ssl_enabled=0, timeout, cid=1 (PDN Context Identifier) server = g.oIni.host_ip # note: host_name could also be used (auto DNS lookup) port = g.oIni.host_port timeout = 10 # secs to wait for response from server cmd = 'AT+SQNHTTPCFG=1,"{0}",{1},0,"","",0,{2},1'.format(server, port, timeout) a = at(cmd,'OK') if len(a) == 0: log("error initializing modem") set_sms_mode() enable_modem() except Exception as ex: sys.print_exception(ex) ##################################################################### # Initialize Uart Connected To LTE Modem ##################################################################### def init_uart(): global uart # pins=(TXD, RXD, RTS, CTS) # FiPy: pins=('P20', 'P18', 'P19', 'P17') # GPy: pins=('P5', 'P98', 'P7', 'P99') log("initializing UART") # rx_buffer_size: must be > 127, 200000 works, 250000 crashes, http_post seems to work with any size uart = UART(1, baudrate=921600, bits=8, parity=None, stop=1, timeout_chars=2, pins=('P5', 'P98', 'P7', 'P99'), rx_buffer_size=512) ##################################################################### # Is Cellular Modem Attached To Packet Domain Service ##################################################################### def isattached(): # returns: True/False try: a = at('AT+CGATT?','+CGATT:') # example response: ['+CGATT: 0', 'OK'] if len(a) == 0: return False if a[0][8] == '1': return True except Exception as ex: sys.print_exception(ex) return False ##################################################################### # Reset Modem ##################################################################### def reset_modem(): try: if uart == None: init_uart() log("reseting modem...") a = at('AT^RESET','+SHUTDOWN') # generates +SHUTDOWN ... +SYSSTART responses if len(a) == 0: log("error resetting modem - no +SHUTDOWN") a = at_response('+SYSSTART') if len(a) == 0: log("error resetting modem - no +SYSSTART") except Exception as ex: sys.print_exception(ex) ##################################################################### # Check Modem For Restart/Crash ##################################################################### def restarted(): # return True if modem has crashed / restarted try: # if CFUN != 1, modem has crashed / restarted cfun = get_cfun() if cfun != 1: return True # if SMS text mode is reset to zero, modem has crashed / restarted mode = get_sms_mode() if mode != 1: return True return False except Exception as ex: sys.print_exception(ex) return True ##################################################################### # Set APN ##################################################################### def set_apn(): try: log("set APN") # ICCID is only valid after registration r = iccid()[0:5] if r == "89014": apn = g.oIni.apn_89014 # Kore AT&T SIM elif r == "89012": apn = g.oIni.apn_89012 # Kore T-Mobile SIM elif r == "89445": apn = g.oIni.apn_89445 # Hologram SIM else: apn = g.oIni.apn_89014 # Kore AT&T SIM g.oIni.apn = apn log("APN: {0}".format(apn)) # try to reduce wear of non-volatile memory a = at('AT+CGDCONT?','+CGDCONT:') # response --> ['+CGDCONT: 1,"IP","10569.mcs",,,,0,0,0,0,0,0', ...] if len(a) > 0: a = a[0].split(",") if (a[0] == '+CGDCONT: 1') and (a[1] == '"IP"') and (a[2] == '"'+apn+'"'): return # apn needs setting cmd = 'AT+CGDCONT=1,"IP","{0}"'.format(apn) a = at(cmd,'OK') if len(a) == 0: log("error setting APN") except Exception as ex: sys.print_exception(ex) ##################################################################### # Set SMS Mode To Text ##################################################################### def set_sms_mode(): try: log("set SMS mode to text") a = at('AT+CMGF=1','OK') # set SMS mode to text if len(a) == 0: log("error setting SMS mode to text") except Exception as ex: sys.print_exception(ex) ##################################################################### # Get Signal Strength ##################################################################### def signal_strength(): # 0-31, 99-unknown try: while True: log("check signal strength") a = at('AT+CSQ','+CSQ:') # example response: ['+CSQ: 18,99', 'OK'] where 18 is signal strength if len(a) == 0: break n = len(a[0]) s = a[0][6:n-1] a = s.split(",") cell_signal = int(a[0]) g.oStatus.cell_signal = cell_signal log("signal strength = {0}".format(cell_signal)) return cell_signal except Exception as ex: sys.print_exception(ex) # unknown signal strength g.oStatus.cell_signal = 99 log("signal strength unknown") return 99 ##################################################################### # Delete Read, Sent, and Unsent SMS Messages ##################################################################### def sms_delete(): try: log("delete all READ, SENT, and UNSENT SMS messages") a = at('AT+CMGD=1,3','OK') if len(a) == 0: log("error trying to delete read, sent, and unsent SMS messages") except Exception as ex: sys.print_exception(ex) ##################################################################### # Delete All SMS Messages ##################################################################### def sms_delete_all(): # doesn't work until some time after attach() is called try: log("delete all SMS messages") a = at('AT+CMGD=1,4','OK') if len(a) == 0: log("error trying to delete all SMS messages") except Exception as ex: sys.print_exception(ex) ##################################################################### # Process SMS Messages ##################################################################### def sms_process(r): # example response: # ['+CMGL: 1,"REC READ","+1xxxxxxxxxx",,"19/01/07,09:16:53-20"', 'Test3', '+CMGL: 2,"REC READ","+1xxxxxxxxxx",,"19/01/07,11:12:30-20"', 'Test4', 'OK'] try: for m in r: if (m[0:5] == "+CMGL"): continue a = m.split("=") cmd = a[0].upper() except Exception as ex: sys.print_exception(ex) ##################################################################### # Read Unread SMS Messages ##################################################################### def sms_read(): # response example: ['+CMGL: 1,"REC UNREAD","+1xxxxxxxxxx",,"19/01/07,09:16:53-20"', 'Test3', 'OK'] log("read unread SMS messages") a = at('AT+CMGL="REC UNREAD"') return a ##################################################################### # Read All SMS Messages ##################################################################### def sms_read_all(): # response example: # ['+CMGL: 1,"REC READ","+1xxxxxxxxxx",,"19/01/07,09:16:53-20"', 'Test3', '+CMGL: 2,"REC READ","+1xxxxxxxxxx",,"19/01/07,11:12:30-20"', 'Test4', 'OK'] log("read all SMS messages") a = at('AT+CMGL="ALL"') return a ##################################################################### # Wait For N Received Chars ##################################################################### def wait_rx(n, timeout=10000): # timeout in msecs t0 = ticks_ms() while True: if uart.any() >= n: break t1 = ticks_ms() tx = ticks_diff(t1, t0) if tx > timeout: return False sleep_ms(10) return TrueHere's the code being used. The code shown is incomplete since it is application dependent. I'm somewhat disappointed in the necessary placement of the modem.apn_set() call. This may change in the future.
# COM.PY Module import globals as g import modem from poll import poll import pycom import sys import uos import util from util import log from utime import sleep_ms import wifi ##################################################################### # Communication Task ##################################################################### def com_task(): try: sleep_ms(500) log("com_task: started") wifi.deinit() modem.reset_modem() modem.init_modem() g.poll_host = True # triggers poll at startup except Exception as ex: sys.print_exception(ex) while True: try: if modem_ready(): if ready_to_poll(): modem.set_apn() transmit() sleep_ms(2000) except Exception as ex: sys.print_exception(ex) ##################################################################### # Modem Ready ##################################################################### def modem_ready(): try: log("** check for modem ready") # get modem signal strength g.oStatus.cell_signal = modem.signal_strength() # check modem registration r = modem.check_registration() if (r == 1) or (r == 5): # registered g.com_secs = 0 # handle SMS messages r = modem.sms_read() if len(r) > 1: modem.sms_process(r) modem.sms_delete() return True reset = False if g.com_secs > (10*60): # unable to attach in 10 mins, try resetting modem reset = True if modem.restarted(): # modem has crashed / restarted, try resetting modem reset = True if reset: modem.reset_modem() modem.init_modem() g.com_secs = 0 return False except Exception as ex: sys.print_exception(ex) return False -
I seemed to have hit a major problem using the AT+SQNHTTPSND modem command. Although the manual does not state this, the maximum size of the data posted is apparently 1500 bytes.
If anyone has ideas of how to increase this, please let me know. Unfortunately my app requires posting messages over 3000 bytes.
Disappointed...
Hello! It looks like you're interested in this conversation, but you don't have an account yet.
Getting fed up of having to scroll through the same posts each visit? When you register for an account, you'll always come back to exactly where you were before, and choose to be notified of new replies (either via email, or push notification). You'll also be able to save bookmarks and upvote posts to show your appreciation to other community members.
With your input, this post could be even better 💗
Register Login