-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathsim900.py
159 lines (123 loc) · 4.38 KB
/
sim900.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
import re
import sqlite3
from time import sleep
class TextMsg(object):
"""
Represents a text message with some meta data
Args:
phone_number: Example format format: +16505033896
timestamp: Example format: 14/05/30,00:13:34-32
message: Text message body with CRLF removed
"""
def __init__(self, phone_number, timestamp, message):
self.phone_number = phone_number
self.timestamp = timestamp
self.message = message.strip()
def __eq__(self, other):
return (isinstance(other, self.__class__)
and self.__dict__ == other.__dict__)
def __str__(self):
return ', '.join([self.phone_number, self.timestamp, self.message])
class Sim900(object):
"""
Sends commands and read input from Sim900 shield.
Note that if you are sending commands to an Arduino,
then the Arduino needs to be loaded with a sketch that
proxies commands to the shield and also forwards the
response through serial.
With the pcDuino, this class communicates directly
with the shield.
"""
CRLF = "\r\n"
CTRL_Z = chr(26)
DELAY_AFTER_READ = 0.1
def __init__(self, serial, delay=0.1):
self.serial = serial
self.delay = delay
def send_cmd(self, cmd, delay=None):
"""
Sends AT commands to Sim900 shield. A CRLF
is automatically added to the command.
Args:
cmd: AT Command to send to shield
delay: Custom delay after sending command. Default is 0.1s
"""
self.serial.write(cmd)
self.serial.write(Sim900.CRLF)
sleep(delay if delay is not None else self.delay)
def available(self):
return self.serial.inWaiting()
def read(self, num_chars=1):
return self.serial.read(num_chars)
def read_available(self):
return self.serial.read(self.available())
def read_all(self):
"""
Attempts to read all incoming input even if the
baud rate is very slow (ie 4800 bps) and only returns
if no change is encountered.
"""
msg = ""
prev_len = 0
curr_len = 0
while True:
prev_len = curr_len
while self.available() != 0:
msg += self.read_available()
curr_len = len(msg)
sleep(self.DELAY_AFTER_READ)
if prev_len == curr_len:
break
return msg
class SMSReader(object):
"""
Listens for incoming SMS text message and extracts
header and message for further processing.
Example format:
+CMT: "+16505033896","","14/05/30,00:13:34-32"<CRLF>
This is the text message body!<CRLF>
Note that the GSM shield can be set to include other metadata
in the +CMT header.
"""
DATA_BEGIN = "+CMT"
DATA_DELIM = "\r\n"
NOT_FOUND = -1
MSG_FORMAT = "\+CMT: \"(\+\d{11})\",\"\",\"(\d{2}\/\d{2}\/\d{2},\d{2}:\d{2}:\d{2}[\-\+]\d{2})\"\r\n(.*)\r\n"
def __init__(self, sim900):
self.sim900 = sim900
self.sms_regex = re.compile(self.MSG_FORMAT)
def init_reader(self):
"""
Makes sure Sim900 shield is set to listen
for incoming SMS text message in text mode.
For the PcDuino, make sure to set the baudrate to
115200. Otherwise, data will be garbled.
This step can be skipped if you are sure that the
shield is set correctly.
For instance if you are proxying commands/responses
through an Arduino, the Arduino sketch may already do
this.
Returns:
Sim900 response to commands.
"""
self.sim900.send_cmd("AT+CMGF=1")
self.sim900.send_cmd("AT+CNMI=2,2,0,0,0")
return self.sim900.read_all()
def listen(self):
"""
Listens for incoming SMS text message with +CMT response code.
Returns:
If SMS text message is found, TextMsg is returned
If message not found, then None is returned
"""
msg = self.sim900.read_all()
return self.extract_sms(msg)
def extract_sms(self, msg):
"""
Extracts SMS text message just in case the message includes
gibberish before or after.
Returns:
TextMsg object or None if content is not in the correct format
"""
result = self.sms_regex.search(msg)
return TextMsg(*result.groups()) if result else None