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
|
# pylint: disable=line-too-long
"""Handle AIS VD[MO] messages without extra metadata.
An example single line message:
!AIVDM,1,1,,A,14VIk0002sMM04vE>V9jGimn08RP,0*0D
An example two line message:
!AIVDM,2,1,2,B,5KNp?6@00000=Phb220u1@Tlv0TV22222222220N0h:22400000000000000,0*2A
!AIVDM,2,2,2,B,00000000000,2*25'
The Parse class calculates the checksum and compares that to expected
checksum at the end of the message. It then uses the VDM_RE regular
expression to create a dictionary for a single line. These
dictionaries are used by the BareQueue class to construct a queue of
complete AIS messages. It also passes non AIS VDM messages through.
Invalid AIS messages are dropped.
The non AIS VDM messages have a simpler data structure like this:
{
'line_nums': [42],
'lines': ['A line of text']
}
A single line AIS VDM message adds a 'decoded' dictionary with the
unpacked message and a list of matches for each of the lines that came
in. Additionally, the queue adds the md5 hash of the body of the
message to assist with the process of looking for duplicate messages.
{
'line_nums': [1],
'lines': ['!SAVDM,1,1,,B,K8VSqb9LdU28WP8P,0*7B'],
'decoded': {
'md5': '6e1a4872825054e91ee7cfcfb9cc87e0',
'id': 27,
'mmsi': 577305000,
# SNIP
},
'matches': [{
'body': 'K8VSqb9LdU28WP8P',
'fill_bits': '0',
'sen_num': '1',
'sen_tot': '1',
'vdm_type': 'VDM',
'talker': 'SA',
'chan': 'B',
'seq_id': None,
'checksum': '7B'}]
}
TODO(schwehr): Make sure that multi-line messages are coming in on the
same channel.
See Also:
http://catb.org/gpsd/AIVDM.html
http://www.itu.int/rec/R-REC-M.1371/en
"""
import hashlib
import logging
import re
import six
import six.moves.queue as Queue
import ais
from ais import nmea
from ais import nmea_messages
from ais import util
logger = logging.getLogger('libais')
# Orbcomm sometimes leaves out the channel.
# TAG BLOCKS use "sentence" as the regex group name. Use sen here to
# avoid collision.
VDM_RE_STR = r"""(?P<vdm>
!(?P<talker>[A-Z][A-Z])(?P<vdm_type>VD[MO])
,(?P<sen_tot>\d) # Total number of sentences.
,(?P<sen_num>\d) # Current sentence number. Starts from 1.
,(?P<seq_id>[0-9])? # Receiver sequence number.
,(?P<chan>[AB])? # VHF channel.
,(?P<body>[;:=@a-zA-Z0-9<>\?\'\`]*)
,(?P<fill_bits>\d)\*(?P<checksum>[0-9A-F][0-9A-F])
) # end vdm
"""
VDM_RE = re.compile(VDM_RE_STR, re.VERBOSE)
NUMERIC_FIELDS = ('fill_bits', 'sen_num', 'sen_tot', 'seq_id')
def VdmLines(lines):
"""Yield only the lines that contain AIS messages.
Args:
lines: An iterable series of strings.
Yields:
Lines that look like they probably contain valid VDM/VDO messages.
"""
for line in lines:
line = line.rstrip()
if nmea.ID_BARE_VDM_RE.match(line):
yield line
def Parse(data):
"""Unpack a NMEA VDM AIS message line(s)."""
if not isinstance(data, six.string_types):
raise NotImplementedError
try:
result = VDM_RE.search(data).groupdict()
except AttributeError:
return
result.update({k: util.MaybeToNumber(v)
for k, v in six.iteritems(result) if k in NUMERIC_FIELDS})
actual = nmea.Checksum(result['vdm'])
expected = result['checksum']
if actual != expected:
return
return result
class BareQueue(Queue.Queue):
"""Build complete AIS messages and create a queue of decoded content.
Manages parsing a stream of NMEA AIS VDM messages. Single line
messages are decoded and added to the queue. Parts of multi-line VDM
messages are stored until a complete message is available. The
bodies of each part are joined with the fill_bit count from the final
sentence are decoded and added to the queue.
"""
def __init__(self):
self.groups = {}
self.line_num = 0
Queue.Queue.__init__(self)
def put(self, line, line_num=None):
"""Add a line of NMEA or raw text to the queue."""
if line_num is not None:
self.line_num = line_num
else:
self.line_num += 1
line_num = self.line_num
line = line.rstrip()
match = Parse(line)
if not match:
logger.info('No VDM match for line: %d, %s', line_num, line)
msg = {
'line_nums': [line_num],
'lines': [line]}
decoded = nmea_messages.DecodeLine(line)
if decoded:
msg['decoded'] = decoded
else:
logger.info('No NMEA match for line: %d, %s', line_num, line)
Queue.Queue.put(self, msg)
return
sentence_total = int(match['sen_tot'])
if sentence_total == 1:
body = match['body']
fill_bits = int(match['fill_bits'])
try:
decoded = ais.decode(body, fill_bits)
except ais.DecodeError as error:
logger.error(
'Unable to decode message: %s\n %d %s', error, line_num, line)
return
decoded['md5'] = hashlib.md5(body.encode('utf-8')).hexdigest()
Queue.Queue.put(self, {
'line_nums': [line_num],
'lines': [line],
'decoded': decoded,
'matches': [match]
})
return
sentence_num = int(match['sen_num'])
group_id = int(match['seq_id'])
if sentence_num == 1:
if group_id in self.groups:
logger.error('Incomplete message overwritten by new start. '
'Dropped:\n %s', self.groups[group_id])
self.groups[group_id] = {
'line_nums': [line_num],
'lines': [line],
'matches': [match]
}
return
if group_id not in self.groups:
logger.error('Do not have the prior lines in group_id %d. '
'Dropping: \n %s', group_id, line)
return
entry = self.groups[group_id]
if len(entry['lines']) != sentence_num - 1:
logger.error('Out of sequence message. Dropping: %d != %d \n %s',
len(entry['lines']), sentence_num - 1, line)
return
entry['lines'].append(line)
entry['matches'].append(match)
entry['line_nums'].append(line_num)
if sentence_num != sentence_total:
# Nothing more to do in the middle of a sequence of sentences.
return
body = ''.join([match['body'] for match in entry['matches']])
fill_bits = int(entry['matches'][-1]['fill_bits'])
try:
decoded = ais.decode(body, fill_bits)
except ais.DecodeError as error:
logger.error(
'Unable to decode message: %s\n%s', error, entry)
return
decoded['md5'] = hashlib.md5(body.encode('utf-8')).hexdigest()
entry['decoded'] = decoded
# Found the final message in a group.
Queue.Queue.put(self, entry)
self.groups.pop(group_id)
|
