forked from faucamp/python-gsmmodem
-
Notifications
You must be signed in to change notification settings - Fork 105
/
modem.py
1746 lines (1542 loc) · 79.2 KB
/
modem.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python
""" High-level API classes for an attached GSM modem """
import sys, re, logging, weakref, time, threading, abc, codecs
from datetime import datetime
from time import sleep
from .serial_comms import SerialComms
from .exceptions import CommandError, InvalidStateException, CmeError, CmsError, InterruptedException, TimeoutException, PinRequiredError, IncorrectPinError, SmscNumberUnknownError
from .pdu import encodeSmsSubmitPdu, decodeSmsPdu, encodeGsm7, encodeTextMode
from .util import SimpleOffsetTzInfo, lineStartingWith, allLinesMatchingPattern, parseTextModeTimeStr, removeAtPrefix
#from . import compat # For Python 2.6 compatibility
from gsmmodem.util import lineMatching
from gsmmodem.exceptions import EncodingError
PYTHON_VERSION = sys.version_info[0]
CTRLZ = '\x1a'
TERMINATOR = '\r'
if PYTHON_VERSION >= 3:
xrange = range
dictValuesIter = dict.values
dictItemsIter = dict.items
else: #pragma: no cover
dictValuesIter = dict.itervalues
dictItemsIter = dict.iteritems
class Sms(object):
""" Abstract SMS message base class """
__metaclass__ = abc.ABCMeta
# Some constants to ease handling SMS statuses
STATUS_RECEIVED_UNREAD = 0
STATUS_RECEIVED_READ = 1
STATUS_STORED_UNSENT = 2
STATUS_STORED_SENT = 3
STATUS_ALL = 4
# ...and a handy converter for text mode statuses
TEXT_MODE_STATUS_MAP = {'REC UNREAD': STATUS_RECEIVED_UNREAD,
'REC READ': STATUS_RECEIVED_READ,
'STO UNSENT': STATUS_STORED_UNSENT,
'STO SENT': STATUS_STORED_SENT,
'ALL': STATUS_ALL}
def __init__(self, number, text, smsc=None):
self.number = number
self.text = text
self.smsc = smsc
class ReceivedSms(Sms):
""" An SMS message that has been received (MT) """
def __init__(self, gsmModem, status, number, time, text, smsc=None, udh=[], index=None):
super(ReceivedSms, self).__init__(number, text, smsc)
self._gsmModem = weakref.proxy(gsmModem)
self.status = status
self.time = time
self.udh = udh
self.index = index
def reply(self, message):
""" Convenience method that sends a reply SMS to the sender of this message """
return self._gsmModem.sendSms(self.number, message)
def sendSms(self, dnumber, message):
""" Convenience method that sends a SMS to someone else """
return self._gsmModem.sendSms(dnumber, message)
def getModem(self):
""" Convenience method that returns the gsm modem instance """
return self._gsmModem
class SentSms(Sms):
""" An SMS message that has been sent (MO) """
ENROUTE = 0 # Status indicating message is still enroute to destination
DELIVERED = 1 # Status indicating message has been received by destination handset
FAILED = 2 # Status indicating message delivery has failed
def __init__(self, number, text, reference, smsc=None):
super(SentSms, self).__init__(number, text, smsc)
self.report = None # Status report for this SMS (StatusReport object)
self.reference = reference
@property
def status(self):
""" Status of this SMS. Can be ENROUTE, DELIVERED or FAILED
The actual status report object may be accessed via the 'report' attribute
if status is 'DELIVERED' or 'FAILED'
"""
if self.report == None:
return SentSms.ENROUTE
else:
return SentSms.DELIVERED if self.report.deliveryStatus == StatusReport.DELIVERED else SentSms.FAILED
class StatusReport(Sms):
""" An SMS status/delivery report
Note: the 'status' attribute of this class refers to this status report SM's status (whether
it has been read, etc). To find the status of the message that caused this status report,
use the 'deliveryStatus' attribute.
"""
DELIVERED = 0 # SMS delivery status: delivery successful
FAILED = 68 # SMS delivery status: delivery failed
def __init__(self, gsmModem, status, reference, number, timeSent, timeFinalized, deliveryStatus, smsc=None):
super(StatusReport, self).__init__(number, None, smsc)
self._gsmModem = weakref.proxy(gsmModem)
self.status = status
self.reference = reference
self.timeSent = timeSent
self.timeFinalized = timeFinalized
self.deliveryStatus = deliveryStatus
class GsmModem(SerialComms):
""" Main class for interacting with an attached GSM modem """
log = logging.getLogger('gsmmodem.modem.GsmModem')
# Used for parsing AT command errors
CM_ERROR_REGEX = re.compile('^\+(CM[ES]) ERROR: (\d+)$')
# Used for parsing signal strength query responses
CSQ_REGEX = re.compile('^\+CSQ:\s*(\d+),')
# Used for parsing caller ID announcements for incoming calls. Group 1 is the number
CLIP_REGEX = re.compile('^\+CLIP:\s*"\+{0,1}(\d+)",(\d+).*$')
# Used for parsing own number. Group 1 is the number
CNUM_REGEX = re.compile('^\+CNUM:\s*".*?","(\+{0,1}\d+)",(\d+).*$')
# Used for parsing new SMS message indications
CMTI_REGEX = re.compile('^\+CMTI:\s*"([^"]+)",\s*(\d+)$')
# Used for parsing SMS message reads (text mode)
CMGR_SM_DELIVER_REGEX_TEXT = None
# Used for parsing SMS status report message reads (text mode)
CMGR_SM_REPORT_REGEXT_TEXT = None
# Used for parsing SMS message reads (PDU mode)
CMGR_REGEX_PDU = None
# Used for parsing USSD event notifications
CUSD_REGEX = re.compile('\+CUSD:\s*(\d),\s*"(.*?)",\s*(\d+)', re.DOTALL)
# Used for parsing SMS status reports
CDSI_REGEX = re.compile('\+CDSI:\s*"([^"]+)",(\d+)$')
CDS_REGEX = re.compile('\+CDS:\s*([0-9]+)"$')
def __init__(self, port, baudrate=115200, incomingCallCallbackFunc=None, smsReceivedCallbackFunc=None, smsStatusReportCallback=None, requestDelivery=True, AT_CNMI="", *a, **kw):
super(GsmModem, self).__init__(port, baudrate, notifyCallbackFunc=self._handleModemNotification, *a, **kw)
self.incomingCallCallback = incomingCallCallbackFunc or self._placeholderCallback
self.smsReceivedCallback = smsReceivedCallbackFunc or self._placeholderCallback
self.smsStatusReportCallback = smsStatusReportCallback or self._placeholderCallback
self.requestDelivery = requestDelivery
self.AT_CNMI = AT_CNMI or "2,1,0,2"
# Flag indicating whether caller ID for incoming call notification has been set up
self._callingLineIdentification = False
# Flag indicating whether incoming call notifications have extended information
self._extendedIncomingCallIndication = False
# Current active calls (ringing and/or answered), key is the unique call ID (not the remote number)
self.activeCalls = {}
# Dict containing sent SMS messages (for auto-tracking their delivery status)
self.sentSms = weakref.WeakValueDictionary()
self._ussdSessionEvent = None # threading.Event
self._ussdResponse = None # gsmmodem.modem.Ussd
self._smsStatusReportEvent = None # threading.Event
self._dialEvent = None # threading.Event
self._dialResponse = None # gsmmodem.modem.Call
self._waitForAtdResponse = True # Flag that controls if we should wait for an immediate response to ATD, or not
self._waitForCallInitUpdate = True # Flag that controls if we should wait for a ATD "call initiated" message
self._callStatusUpdates = [] # populated during connect() - contains regexes and handlers for detecting/handling call status updates
self._mustPollCallStatus = False # whether or not the modem must be polled for outgoing call status updates
self._pollCallStatusRegex = None # Regular expression used when polling outgoing call status
self._writeWait = 0 # Time (in seconds to wait after writing a command (adjusted when 515 errors are detected)
self._smsTextMode = False # Storage variable for the smsTextMode property
self._gsmBusy = 0 # Storage variable for the GSMBUSY property
self._smscNumber = None # Default SMSC number
self._smsRef = 0 # Sent SMS reference counter
self._smsMemReadDelete = None # Preferred message storage memory for reads/deletes (<mem1> parameter used for +CPMS)
self._smsMemWrite = None # Preferred message storage memory for writes (<mem2> parameter used for +CPMS)
self._smsReadSupported = True # Whether or not reading SMS messages is supported via AT commands
self._smsEncoding = 'GSM' # Default SMS encoding
self._smsSupportedEncodingNames = None # List of available encoding names
self._commands = None # List of supported AT commands
#Pool of detected DTMF
self.dtmfpool = []
def connect(self, pin=None, waitingForModemToStartInSeconds=0):
""" Opens the port and initializes the modem and SIM card
:param pin: The SIM card PIN code, if any
:type pin: str
:raise PinRequiredError: if the SIM card requires a PIN but none was provided
:raise IncorrectPinError: if the specified PIN is incorrect
"""
self.log.info('Connecting to modem on port %s at %dbps', self.port, self.baudrate)
super(GsmModem, self).connect()
if waitingForModemToStartInSeconds > 0:
while waitingForModemToStartInSeconds > 0:
try:
self.write('AT', waitForResponse=True, timeout=0.5)
break
except TimeoutException:
waitingForModemToStartInSeconds -= 0.5
# Send some initialization commands to the modem
try:
self.write('ATZ') # reset configuration
except CommandError:
# Some modems require a SIM PIN at this stage already; unlock it now
# Attempt to enable detailed error messages (to catch incorrect PIN error)
# but ignore if it fails
self.write('AT+CMEE=1', parseError=False)
self._unlockSim(pin)
pinCheckComplete = True
self.write('ATZ') # reset configuration
else:
pinCheckComplete = False
self.write('ATE0') # echo off
try:
cfun = lineStartingWith('+CFUN:', self.write('AT+CFUN?'))[7:] # example response: +CFUN: 1 or +CFUN: 1,0
cfun = int(cfun.split(",")[0])
if cfun != 1:
self.write('AT+CFUN=1')
except CommandError:
pass # just ignore if the +CFUN command isn't supported
self.write('AT+CMEE=1') # enable detailed error messages (even if it has already been set - ATZ may reset this)
if not pinCheckComplete:
self._unlockSim(pin)
# Get list of supported commands from modem
commands = self.supportedCommands
self._commands = commands
# Device-specific settings
callUpdateTableHint = 0 # unknown modem
enableWind = False
if commands != None:
if '^CVOICE' in commands:
self.write('AT^CVOICE=0', parseError=False) # Enable voice calls
if '+VTS' in commands: # Check for DTMF sending support
Call.dtmfSupport = True
elif '^DTMF' in commands:
# Huawei modems use ^DTMF to send DTMF tones
callUpdateTableHint = 1 # Huawei
if '^USSDMODE' in commands:
# Enable Huawei text-mode USSD
self.write('AT^USSDMODE=0', parseError=False)
if '+WIND' in commands:
callUpdateTableHint = 2 # Wavecom
enableWind = True
elif '+ZPAS' in commands:
callUpdateTableHint = 3 # ZTE
else:
# Try to enable general notifications on Wavecom-like device
enableWind = True
if enableWind:
try:
wind = lineStartingWith('+WIND:', self.write('AT+WIND?')) # Check current WIND value; example response: +WIND: 63
except CommandError:
# Modem does not support +WIND notifications. See if we can detect other known call update notifications
pass
else:
# Enable notifications for call setup, hangup, etc
if int(wind[7:]) != 50:
self.write('AT+WIND=50')
callUpdateTableHint = 2 # Wavecom
# Attempt to identify modem type directly (if not already) - for outgoing call status updates
if callUpdateTableHint == 0:
if 'simcom' in self.manufacturer.lower() : #simcom modems support DTMF and don't support AT+CLAC
Call.dtmfSupport = True
try:
self.write('AT+DDET=1') # enable detect incoming DTMF
except CommandError:
# simcom 7000E for example doesn't support the DDET command
Call.dtmfSupport = False
if self.manufacturer.lower() == 'huawei':
callUpdateTableHint = 1 # huawei
else:
# See if this is a ZTE modem that has not yet been identified based on supported commands
try:
self.write('AT+ZPAS?')
except CommandError:
pass # Not a ZTE modem
else:
callUpdateTableHint = 3 # ZTE
# Load outgoing call status updates based on identified modem features
if callUpdateTableHint == 1:
# Use Hauwei's ^NOTIFICATIONs
self.log.info('Loading Huawei call state update table')
self._callStatusUpdates = ((re.compile('^\^ORIG:(\d),(\d)$'), self._handleCallInitiated),
(re.compile('^\^CONN:(\d),(\d)$'), self._handleCallAnswered),
(re.compile('^\^CEND:(\d),(\d+),(\d)+,(\d)+$'), self._handleCallEnded))
self._mustPollCallStatus = False
# Huawei modems use ^DTMF to send DTMF tones; use that instead
Call.DTMF_COMMAND_BASE = '^DTMF={cid},'
Call.dtmfSupport = True
elif callUpdateTableHint == 2:
# Wavecom modem: +WIND notifications supported
self.log.info('Loading Wavecom call state update table')
self._callStatusUpdates = ((re.compile('^\+WIND: 5,(\d)$'), self._handleCallInitiated),
(re.compile('^OK$'), self._handleCallAnswered),
(re.compile('^\+WIND: 6,(\d)$'), self._handleCallEnded))
self._waitForAtdResponse = False # Wavecom modems return OK only when the call is answered
self._mustPollCallStatus = False
if commands == None: # older modem, assume it has standard DTMF support
Call.dtmfSupport = True
elif callUpdateTableHint == 3: # ZTE
# Use ZTE notifications ("CONNECT"/"HANGUP", but no "call initiated" notification)
self.log.info('Loading ZTE call state update table')
self._callStatusUpdates = ((re.compile('^CONNECT$'), self._handleCallAnswered),
(re.compile('^HANGUP:\s*(\d+)$'), self._handleCallEnded),
(re.compile('^OK$'), self._handleCallRejected))
self._waitForAtdResponse = False # ZTE modems do not return an immediate OK only when the call is answered
self._mustPollCallStatus = False
self._waitForCallInitUpdate = False # ZTE modems do not provide "call initiated" updates
if commands == None: # ZTE uses standard +VTS for DTMF
Call.dtmfSupport = True
else:
# Unknown modem - we do not know what its call updates look like. Use polling instead
self.log.info('Unknown/generic modem type - will use polling for call state updates')
self._mustPollCallStatus = True
self._pollCallStatusRegex = re.compile('^\+CLCC:\s+(\d+),(\d),(\d),(\d),([^,]),"([^,]*)",(\d+)$')
self._waitForAtdResponse = True # Most modems return OK immediately after issuing ATD
# General meta-information setup
self.write('AT+COPS=3,0', parseError=False) # Use long alphanumeric name format
# SMS setup
self.write('AT+CMGF={0}'.format(1 if self.smsTextMode else 0)) # Switch to text or PDU mode for SMS messages
self._compileSmsRegexes()
if self._smscNumber != None:
self.write('AT+CSCA="{0}"'.format(self._smscNumber)) # Set default SMSC number
currentSmscNumber = self._smscNumber
else:
currentSmscNumber = self.smsc
# Some modems delete the SMSC number when setting text-mode SMS parameters; preserve it if needed
if currentSmscNumber != None:
self._smscNumber = None # clear cache
if self.requestDelivery:
self.write('AT+CSMP=49,167,0,0', parseError=False) # Enable delivery reports
else:
self.write('AT+CSMP=17,167,0,0', parseError=False) # Not enable delivery reports
# ...check SMSC again to ensure it did not change
if currentSmscNumber != None and self.smsc != currentSmscNumber:
self.smsc = currentSmscNumber
# Set message storage, but first check what the modem supports - example response: +CPMS: (("SM","BM","SR"),("SM"))
try:
cpmsLine = lineStartingWith('+CPMS', self.write('AT+CPMS=?'))
except CommandError:
# Modem does not support AT+CPMS; SMS reading unavailable
self._smsReadSupported = False
self.log.warning('SMS preferred message storage query not supported by modem. SMS reading unavailable.')
else:
cpmsSupport = cpmsLine.split(' ', 1)[1].split('),(')
# Do a sanity check on the memory types returned - Nokia S60 devices return empty strings, for example
for memItem in cpmsSupport:
if len(memItem) == 0:
# No support for reading stored SMS via AT commands - probably a Nokia S60
self._smsReadSupported = False
self.log.warning('Invalid SMS message storage support returned by modem. SMS reading unavailable. Response was: "%s"', cpmsLine)
break
else:
# Suppported memory types look fine, continue
preferredMemoryTypes = ('"ME"', '"SM"', '"SR"')
cpmsItems = [''] * len(cpmsSupport)
for i in xrange(len(cpmsSupport)):
for memType in preferredMemoryTypes:
if memType in cpmsSupport[i]:
if i == 0:
self._smsMemReadDelete = memType
cpmsItems[i] = memType
break
self.write('AT+CPMS={0}'.format(','.join(cpmsItems))) # Set message storage
del cpmsSupport
del cpmsLine
if self._smsReadSupported and (self.smsReceivedCallback or self.smsStatusReportCallback):
try:
self.write('AT+CNMI=' + self.AT_CNMI) # Set message notifications
except CommandError:
try:
self.write('AT+CNMI=2,1,0,1,0') # Set message notifications, using TE for delivery reports <ds>
except CommandError:
# Message notifications not supported
self._smsReadSupported = False
self.log.warning('Incoming SMS notifications not supported by modem. SMS receiving unavailable.')
# Incoming call notification setup
try:
self.write('AT+CLIP=1') # Enable calling line identification presentation
except CommandError as clipError:
self._callingLineIdentification = False
self.log.warning('Incoming call calling line identification (caller ID) not supported by modem. Error: {0}'.format(clipError))
else:
self._callingLineIdentification = True
try:
self.write('AT+CRC=1') # Enable extended format of incoming indication (optional)
except CommandError as crcError:
self._extendedIncomingCallIndication = False
self.log.warning('Extended format incoming call indication not supported by modem. Error: {0}'.format(crcError))
else:
self._extendedIncomingCallIndication = True
# Call control setup
self.write('AT+CVHU=0', parseError=False) # Enable call hang-up with ATH command (ignore if command not supported)
def _unlockSim(self, pin):
""" Unlocks the SIM card using the specified PIN (if necessary, else does nothing) """
# Unlock the SIM card if needed
try:
cpinResponse = lineStartingWith('+CPIN', self.write('AT+CPIN?', timeout=15))
except TimeoutException as timeout:
# Wavecom modems do not end +CPIN responses with "OK" (github issue #19) - see if just the +CPIN response was returned
if timeout.data != None:
cpinResponse = lineStartingWith('+CPIN', timeout.data)
if cpinResponse == None:
# No useful response read
raise timeout
else:
# Nothing read (real timeout)
raise timeout
if cpinResponse != '+CPIN: READY':
if pin != None:
self.write('AT+CPIN="{0}"'.format(pin))
else:
raise PinRequiredError('AT+CPIN')
def write(self, data, waitForResponse=True, timeout=10, parseError=True, writeTerm=TERMINATOR, expectedResponseTermSeq=None):
""" Write data to the modem.
This method adds the ``\\r\\n`` end-of-line sequence to the data parameter, and
writes it to the modem.
:param data: Command/data to be written to the modem
:type data: str
:param waitForResponse: Whether this method should block and return the response from the modem or not
:type waitForResponse: bool
:param timeout: Maximum amount of time in seconds to wait for a response from the modem
:type timeout: int
:param parseError: If True, a CommandError is raised if the modem responds with an error (otherwise the response is returned as-is)
:type parseError: bool
:param writeTerm: The terminating sequence to append to the written data
:type writeTerm: str
:param expectedResponseTermSeq: The expected terminating sequence that marks the end of the modem's response (defaults to ``\\r\\n``)
:type expectedResponseTermSeq: str
:raise CommandError: if the command returns an error (only if parseError parameter is True)
:raise TimeoutException: if no response to the command was received from the modem
:return: A list containing the response lines from the modem, or None if waitForResponse is False
:rtype: list
"""
self.log.debug('write: %s', data)
responseLines = super(GsmModem, self).write(data + writeTerm, waitForResponse=waitForResponse, timeout=timeout, expectedResponseTermSeq=expectedResponseTermSeq)
if self._writeWait > 0: # Sleep a bit if required (some older modems suffer under load)
time.sleep(self._writeWait)
if waitForResponse:
cmdStatusLine = responseLines[-1]
if parseError:
if 'ERROR' in cmdStatusLine:
cmErrorMatch = self.CM_ERROR_REGEX.match(cmdStatusLine)
if cmErrorMatch:
errorType = cmErrorMatch.group(1)
errorCode = int(cmErrorMatch.group(2))
if errorCode == 515 or errorCode == 14:
# 515 means: "Please wait, init or command processing in progress."
# 14 means "SIM busy"
self._writeWait += 0.2 # Increase waiting period temporarily
# Retry the command after waiting a bit
self.log.debug('Device/SIM busy error detected; self._writeWait adjusted to %fs', self._writeWait)
time.sleep(self._writeWait)
result = self.write(data, waitForResponse, timeout, parseError, writeTerm, expectedResponseTermSeq)
self.log.debug('self_writeWait set to 0.1 because of recovering from device busy (515) error')
if errorCode == 515:
self._writeWait = 0.1 # Set this to something sane for further commands (slow modem)
else:
self._writeWait = 0 # The modem was just waiting for the SIM card
return result
if errorType == 'CME':
raise CmeError(data, int(errorCode))
else: # CMS error
raise CmsError(data, int(errorCode))
else:
raise CommandError(data)
elif cmdStatusLine == 'COMMAND NOT SUPPORT': # Some Huawei modems respond with this for unknown commands
raise CommandError('{} ({})'.format(data,cmdStatusLine))
return responseLines
@property
def signalStrength(self):
""" Checks the modem's cellular network signal strength
:raise CommandError: if an error occurs
:return: The network signal strength as an integer between 0 and 99, or -1 if it is unknown
:rtype: int
"""
csq = self.CSQ_REGEX.match(self.write('AT+CSQ')[0])
if csq:
ss = int(csq.group(1))
return ss if ss != 99 else -1
else:
raise CommandError()
@property
def manufacturer(self):
""" :return: The modem's manufacturer's name """
return self.write('AT+CGMI')[0]
@property
def model(self):
""" :return: The modem's model name """
return self.write('AT+CGMM')[0]
@property
def revision(self):
""" :return: The modem's software revision, or None if not known/supported """
try:
return self.write('AT+CGMR')[0]
except CommandError:
return None
@property
def imei(self):
""" :return: The modem's serial number (IMEI number) """
return self.write('AT+CGSN')[0]
@property
def imsi(self):
""" :return: The IMSI (International Mobile Subscriber Identity) of the SIM card. The PIN may need to be entered before reading the IMSI """
return self.write('AT+CIMI')[0]
@property
def networkName(self):
""" :return: the name of the GSM Network Operator to which the modem is connected """
copsMatch = lineMatching('^\+COPS: (\d),(\d),"(.+)",{0,1}\d*$', self.write('AT+COPS?')) # response format: +COPS: mode,format,"operator_name",x
if copsMatch:
return copsMatch.group(3)
@property
def supportedCommands(self):
""" :return: list of AT commands supported by this modem (without the AT prefix). Returns None if not known """
try:
# AT+CLAC responses differ between modems. Most respond with +CLAC: and then a comma-separated list of commands
# while others simply return each command on a new line, with no +CLAC: prefix
response = self.write('AT+CLAC', timeout=10)
if len(response) == 2: # Single-line response, comma separated
commands = response[0]
if commands.startswith('+CLAC'):
commands = commands[6:] # remove the +CLAC: prefix before splitting
return commands.split(',')
elif len(response) > 2: # Multi-line response
return [removeAtPrefix(cmd.strip()) for cmd in response[:-1]]
else:
self.log.debug('Unhandled +CLAC response: {0}'.format(response))
return None
except (TimeoutException, CommandError):
# Try interactive command recognition
commands = []
checkable_commands = ['^CVOICE', '+VTS', '^DTMF', '^USSDMODE', '+WIND', '+ZPAS', '+CSCS', '+CNUM']
# Check if modem is still alive
try:
response = self.write('AT')
except:
raise TimeoutException
# Check all commands that will by considered
for command in checkable_commands:
try:
# Compose AT command that will read values under specified function
at_command='AT'+command+'=?'
response = self.write(at_command)
# If there are values inside response - add command to the list
commands.append(command)
except:
continue
# Return found commands
if len(commands) == 0:
return None
else:
return commands
@property
def smsTextMode(self):
""" :return: True if the modem is set to use text mode for SMS, False if it is set to use PDU mode """
return self._smsTextMode
@smsTextMode.setter
def smsTextMode(self, textMode):
""" Set to True for the modem to use text mode for SMS, or False for it to use PDU mode """
if textMode != self._smsTextMode:
if self.alive:
self.write('AT+CMGF={0}'.format(1 if textMode else 0))
self._smsTextMode = textMode
self._compileSmsRegexes()
@property
def smsSupportedEncoding(self):
"""
:raise NotImplementedError: If an error occures during AT command response parsing.
:return: List of supported encoding names. """
# Check if command is available
if self._commands == None:
self._commands = self.supportedCommands
if self._commands == None:
self._smsSupportedEncodingNames = []
return self._smsSupportedEncodingNames
if not '+CSCS' in self._commands:
self._smsSupportedEncodingNames = []
return self._smsSupportedEncodingNames
# Get available encoding names
response = self.write('AT+CSCS=?')
# Check response length (should be 2 - list of options and command status)
if len(response) != 2:
self.log.debug('Unhandled +CSCS response: {0}'.format(response))
self._smsSupportedEncodingNames = []
raise NotImplementedError
# Extract encoding names list
try:
enc_list = response[0] # Get the first line
enc_list = enc_list[6:] # Remove '+CSCS: ' prefix
# Extract AT list in format ("str", "str2", "str3")
enc_list = enc_list.split('(')[1]
enc_list = enc_list.split(')')[0]
enc_list = enc_list.split(',')
enc_list = [x.split('"')[1] for x in enc_list]
except:
self.log.debug('Unhandled +CSCS response: {0}'.format(response))
self._smsSupportedEncodingNames = []
raise NotImplementedError
self._smsSupportedEncodingNames = enc_list
return self._smsSupportedEncodingNames
@property
def smsEncoding(self):
""" :return: Encoding name if encoding command is available, else GSM. """
if self._commands == None:
self._commands = self.supportedCommands
if self._commands == None:
return self._smsEncoding
if '+CSCS' in self._commands:
response = self.write('AT+CSCS?')
if len(response) == 2:
encoding = response[0]
if encoding.startswith('+CSCS'):
encoding = encoding[6:].split('"') # remove the +CSCS: prefix before splitting
if len(encoding) == 3:
self._smsEncoding = encoding[1]
else:
self.log.debug('Unhandled +CSCS response: {0}'.format(response))
else:
self.log.debug('Unhandled +CSCS response: {0}'.format(response))
return self._smsEncoding
@smsEncoding.setter
def smsEncoding(self, encoding):
""" Set encoding for SMS inside PDU mode.
:raise CommandError: if unable to set encoding
:raise ValueError: if encoding is not supported by modem
"""
# Check if command is available
if self._commands == None:
self._commands = self.supportedCommands
if self._commands == None:
if encoding != self._smsEncoding:
raise CommandError('Unable to set SMS encoding (no supported commands)')
else:
return
if not '+CSCS' in self._commands:
if encoding != self._smsEncoding:
raise CommandError('Unable to set SMS encoding (+CSCS command not supported)')
else:
return
# Check if command is available
if self._smsSupportedEncodingNames == None:
self.smsSupportedEncoding
# Check if desired encoding is available
if encoding in self._smsSupportedEncodingNames:
# Set encoding
response = self.write('AT+CSCS="{0}"'.format(encoding))
if len(response) == 1:
if response[0].lower() == 'ok':
self._smsEncoding = encoding
return
if encoding != self._smsEncoding:
raise ValueError('Unable to set SMS encoding (enocoding {0} not supported)'.format(encoding))
else:
return
def _setSmsMemory(self, readDelete=None, write=None):
""" Set the current SMS memory to use for read/delete/write operations """
# Switch to the correct memory type if required
if write != None and write != self._smsMemWrite:
readDel = readDelete or self._smsMemReadDelete
self.write('AT+CPMS="{0}","{1}"'.format(readDel, write))
self._smsMemReadDelete = readDel
self._smsMemWrite = write
elif readDelete != None and readDelete != self._smsMemReadDelete:
self.write('AT+CPMS="{0}"'.format(readDelete))
self._smsMemReadDelete = readDelete
def _compileSmsRegexes(self):
""" Compiles regular expression used for parsing SMS messages based on current mode """
if self.smsTextMode:
if self.CMGR_SM_DELIVER_REGEX_TEXT == None:
self.CMGR_SM_DELIVER_REGEX_TEXT = re.compile('^\+CMGR: "([^"]+)","([^"]+)",[^,]*,"([^"]+)"$')
self.CMGR_SM_REPORT_REGEXT_TEXT = re.compile('^\+CMGR: ([^,]*),\d+,(\d+),"{0,1}([^"]*)"{0,1},\d*,"([^"]+)","([^"]+)",(\d+)$')
elif self.CMGR_REGEX_PDU == None:
self.CMGR_REGEX_PDU = re.compile('^\+CMGR:\s*(\d*),\s*"{0,1}([^"]*)"{0,1},\s*(\d+)$')
@property
def gsmBusy(self):
""" :return: Current GSMBUSY state """
try:
response = self.write('AT+GSMBUSY?')
response = response[0] # Get the first line
response = response[10] # Remove '+GSMBUSY: ' prefix
self._gsmBusy = response
except:
pass # If error is related to ME funtionality: +CME ERROR: <error>
return self._gsmBusy
@gsmBusy.setter
def gsmBusy(self, gsmBusy):
""" Sete GSMBUSY state """
if gsmBusy != self._gsmBusy:
if self.alive:
self.write('AT+GSMBUSY="{0}"'.format(gsmBusy))
self._gsmBusy = gsmBusy
@property
def smsc(self):
""" :return: The default SMSC number stored on the SIM card """
if self._smscNumber == None:
try:
readSmsc = self.write('AT+CSCA?')
except SmscNumberUnknownError:
pass # Some modems return a CMS 330 error if the value isn't set
else:
cscaMatch = lineMatching('\+CSCA:\s*"([^,]+)",(\d+)$', readSmsc)
if cscaMatch:
self._smscNumber = cscaMatch.group(1)
return self._smscNumber
@smsc.setter
def smsc(self, smscNumber):
""" Set the default SMSC number to use when sending SMS messages """
if smscNumber != self._smscNumber:
if self.alive:
self.write('AT+CSCA="{0}"'.format(smscNumber))
self._smscNumber = smscNumber
@property
def ownNumber(self):
""" Query subscriber phone number.
It must be stored on SIM by operator.
If is it not stored already, it usually is possible to store the number by user.
:raise TimeoutException: if a timeout was specified and reached
:return: Subscriber SIM phone number. Returns None if not known
:rtype: int
"""
try:
if "+CNUM" in self._commands:
response = self.write('AT+CNUM')
else:
# temporarily switch to "own numbers" phonebook, read position 1 and than switch back
response = self.write('AT+CPBS?')
selected_phonebook = response[0][6:].split('"')[1] # first line, remove the +CSCS: prefix, split, first parameter
if selected_phonebook is not "ON":
self.write('AT+CPBS="ON"')
response = self.write("AT+CPBR=1")
self.write('AT+CPBS="{0}"'.format(selected_phonebook))
if response is "OK": # command is supported, but no number is set
return None
elif len(response) == 2: # OK and phone number. Actual number is in the first line, second parameter, and is placed inside quotation marks
cnumLine = response[0]
cnumMatch = self.CNUM_REGEX.match(cnumLine)
if cnumMatch:
return cnumMatch.group(1)
else:
self.log.debug('Error parse +CNUM response: {0}'.format(response))
return None
elif len(response) > 2: # Multi-line response
self.log.debug('Unhandled +CNUM/+CPBS response: {0}'.format(response))
return None
except (TimeoutException, CommandError):
raise
@ownNumber.setter
def ownNumber(self, phone_number):
actual_phonebook = self.write('AT+CPBS?')
if actual_phonebook is not "ON":
self.write('AT+CPBS="ON"')
self.write('AT+CPBW=1,"' + phone_number + '"')
def waitForNetworkCoverage(self, timeout=None):
""" Block until the modem has GSM network coverage.
This method blocks until the modem is registered with the network
and the signal strength is greater than 0, optionally timing out
if a timeout was specified
:param timeout: Maximum time to wait for network coverage, in seconds
:type timeout: int or float
:raise TimeoutException: if a timeout was specified and reached
:raise InvalidStateException: if the modem is not going to receive network coverage (SIM blocked, etc)
:return: the current signal strength
"""
block = [True]
if timeout != None:
# Set up a timeout mechanism
def _cancelBlock():
block[0] = False
t = threading.Timer(timeout, _cancelBlock)
t.start()
ss = -1
checkCreg = True
while block[0]:
if checkCreg:
cregResult = lineMatching('^\+CREG:\s*(\d),(\d)(,[^,]*,[^,]*)?$', self.write('AT+CREG?', parseError=False)) # example result: +CREG: 0,1
if cregResult:
status = int(cregResult.group(2))
if status in (1, 5):
# 1: registered, home network, 5: registered, roaming
# Now simply check and return network signal strength
checkCreg = False
elif status == 3:
raise InvalidStateException('Network registration denied')
elif status == 0:
raise InvalidStateException('Device not searching for network operator')
else:
# Disable network registration check; only use signal strength
self.log.info('+CREG check disabled due to invalid response or unsupported command')
checkCreg = False
else:
# Check signal strength
ss = self.signalStrength
if ss > 0:
return ss
time.sleep(1)
else:
# If this is reached, the timer task has triggered
raise TimeoutException()
def sendSms(self, destination, text, waitForDeliveryReport=False, deliveryTimeout=15, sendFlash=False):
""" Send an SMS text message
:param destination: the recipient's phone number
:type destination: str
:param text: the message text
:type text: str
:param waitForDeliveryReport: if True, this method blocks until a delivery report is received for the sent message
:type waitForDeliveryReport: boolean
:param deliveryTimeout: the maximum time in seconds to wait for a delivery report (if "waitForDeliveryReport" is True)
:type deliveryTimeout: int or float
:raise CommandError: if an error occurs while attempting to send the message
:raise TimeoutException: if the operation times out
"""
# Check input text to select appropriate mode (text or PDU)
if self.smsTextMode:
try:
encodedText = encodeTextMode(text)
except ValueError:
self.smsTextMode = False
if self.smsTextMode:
# Send SMS via AT commands
self.write('AT+CMGS="{0}"'.format(destination), timeout=5, expectedResponseTermSeq='> ')
result = lineStartingWith('+CMGS:', self.write(text, timeout=35, writeTerm=CTRLZ))
else:
# Check encoding
try:
encodedText = encodeGsm7(text)
except ValueError:
encodedText = None
# Set GSM modem SMS encoding format
# Encode message text and set data coding scheme based on text contents
if encodedText == None:
# Cannot encode text using GSM-7; use UCS2 instead
self.smsEncoding = 'UCS2'
else:
self.smsEncoding = 'GSM'
# Encode text into PDUs
pdus = encodeSmsSubmitPdu(destination, text, reference=self._smsRef, sendFlash=sendFlash)
# Send SMS PDUs via AT commands
for pdu in pdus:
self.write('AT+CMGS={0}'.format(pdu.tpduLength), timeout=5, expectedResponseTermSeq='> ')
result = lineStartingWith('+CMGS:', self.write(str(pdu), timeout=35, writeTerm=CTRLZ)) # example: +CMGS: xx
if result == None:
raise CommandError('Modem did not respond with +CMGS response')
# Keep SMS reference number in order to pair delivery reports with sent message
reference = int(result[7:])
self._smsRef = reference + 1
if self._smsRef > 255:
self._smsRef = 0
# Create sent SMS object for future delivery checks
sms = SentSms(destination, text, reference)
# Add a weak-referenced entry for this SMS (allows us to update the SMS state if a status report is received)
self.sentSms[reference] = sms
if waitForDeliveryReport:
self._smsStatusReportEvent = threading.Event()
if self._smsStatusReportEvent.wait(deliveryTimeout):
self._smsStatusReportEvent = None
else: # Response timed out
self._smsStatusReportEvent = None
raise TimeoutException()
return sms
def sendUssd(self, ussdString, responseTimeout=15):
""" Starts a USSD session by dialing the the specified USSD string, or \
sends the specified string in the existing USSD session (if any)
:param ussdString: The USSD access number to dial
:param responseTimeout: Maximum time to wait a response, in seconds
:raise TimeoutException: if no response is received in time
:return: The USSD response message/session (as a Ussd object)
:rtype: gsmmodem.modem.Ussd
"""
self._ussdSessionEvent = threading.Event()
try:
cusdResponse = self.write('AT+CUSD=1,"{0}",15'.format(ussdString), timeout=responseTimeout) # Should respond with "OK"
except Exception:
self._ussdSessionEvent = None # Cancel the thread sync lock
raise
# Some modems issue the +CUSD response before the acknowledgment "OK" - check for that
if len(cusdResponse) > 1:
cusdResponseFound = lineStartingWith('+CUSD', cusdResponse) != None
if cusdResponseFound:
self._ussdSessionEvent = None # Cancel thread sync lock
return self._parseCusdResponse(cusdResponse)
# Wait for the +CUSD notification message
if self._ussdSessionEvent.wait(responseTimeout):
self._ussdSessionEvent = None
return self._ussdResponse
else: # Response timed out
self._ussdSessionEvent = None
raise TimeoutException()
def checkForwarding(self, querytype, responseTimeout=15):
""" Check forwarding status: 0=Unconditional, 1=Busy, 2=NoReply, 3=NotReach, 4=AllFwd, 5=AllCondFwd
:param querytype: The type of forwarding to check
:return: Status
:rtype: Boolean
"""
try:
queryResponse = self.write('AT+CCFC={0},2'.format(querytype), timeout=responseTimeout) # Should respond with "OK"
except Exception:
raise
print(queryResponse)