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
|
# -*- test-case-name: twisted.test.test_nmea -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""NMEA 0183 implementation
Maintainer: Bob Ippolito
The following NMEA 0183 sentences are currently understood::
GPGGA (fix)
GPGLL (position)
GPRMC (position and time)
GPGSA (active satellites)
The following NMEA 0183 sentences require implementation::
None really, the others aren't generally useful or implemented in most devices anyhow
Other desired features::
- A NMEA 0183 producer to emulate GPS devices (?)
"""
import operator
from twisted.protocols import basic
from twisted.python.compat import reduce
POSFIX_INVALID, POSFIX_SPS, POSFIX_DGPS, POSFIX_PPS = 0, 1, 2, 3
MODE_AUTO, MODE_FORCED = 'A', 'M'
MODE_NOFIX, MODE_2D, MODE_3D = 1, 2, 3
class InvalidSentence(Exception):
pass
class InvalidChecksum(Exception):
pass
class NMEAReceiver(basic.LineReceiver):
"""This parses most common NMEA-0183 messages, presumably from a serial GPS device at 4800 bps
"""
delimiter = '\r\n'
dispatch = {
'GPGGA': 'fix',
'GPGLL': 'position',
'GPGSA': 'activesatellites',
'GPRMC': 'positiontime',
'GPGSV': 'viewsatellites', # not implemented
'GPVTG': 'course', # not implemented
'GPALM': 'almanac', # not implemented
'GPGRS': 'range', # not implemented
'GPGST': 'noise', # not implemented
'GPMSS': 'beacon', # not implemented
'GPZDA': 'time', # not implemented
}
# generally you may miss the beginning of the first message
ignore_invalid_sentence = 1
# checksums shouldn't be invalid
ignore_checksum_mismatch = 0
# ignore unknown sentence types
ignore_unknown_sentencetypes = 0
# do we want to even bother checking to see if it's from the 20th century?
convert_dates_before_y2k = 1
def lineReceived(self, line):
if not line.startswith('$'):
if self.ignore_invalid_sentence:
return
raise InvalidSentence("%r does not begin with $" % (line,))
# message is everything between $ and *, checksum is xor of all ASCII values of the message
strmessage, checksum = line[1:].strip().split('*')
message = strmessage.split(',')
sentencetype, message = message[0], message[1:]
dispatch = self.dispatch.get(sentencetype, None)
if (not dispatch) and (not self.ignore_unknown_sentencetypes):
raise InvalidSentence("sentencetype %r" % (sentencetype,))
if not self.ignore_checksum_mismatch:
checksum, calculated_checksum = int(checksum, 16), reduce(operator.xor, map(ord, strmessage))
if checksum != calculated_checksum:
raise InvalidChecksum("Given 0x%02X != 0x%02X" % (checksum, calculated_checksum))
handler = getattr(self, "handle_%s" % dispatch, None)
decoder = getattr(self, "decode_%s" % dispatch, None)
if not (dispatch and handler and decoder):
# missing dispatch, handler, or decoder
return
# return handler(*decoder(*message))
try:
decoded = decoder(*message)
except Exception, e:
raise InvalidSentence("%r is not a valid %s (%s) sentence" % (line, sentencetype, dispatch))
return handler(*decoded)
def decode_position(self, latitude, ns, longitude, ew, utc, status):
latitude, longitude = self._decode_latlon(latitude, ns, longitude, ew)
utc = self._decode_utc(utc)
if status == 'A':
status = 1
else:
status = 0
return (
latitude,
longitude,
utc,
status,
)
def decode_positiontime(self, utc, status, latitude, ns, longitude, ew, speed, course, utcdate, magvar, magdir):
utc = self._decode_utc(utc)
latitude, longitude = self._decode_latlon(latitude, ns, longitude, ew)
if speed != '':
speed = float(speed)
else:
speed = None
if course != '':
course = float(course)
else:
course = None
utcdate = 2000+int(utcdate[4:6]), int(utcdate[2:4]), int(utcdate[0:2])
if self.convert_dates_before_y2k and utcdate[0] > 2073:
# GPS was invented by the US DoD in 1973, but NMEA uses 2 digit year.
# Highly unlikely that we'll be using NMEA or this twisted module in 70 years,
# but remotely possible that you'll be using it to play back data from the 20th century.
utcdate = (utcdate[0] - 100, utcdate[1], utcdate[2])
if magvar != '':
magvar = float(magvar)
if magdir == 'W':
magvar = -magvar
else:
magvar = None
return (
latitude,
longitude,
speed,
course,
# UTC seconds past utcdate
utc,
# UTC (year, month, day)
utcdate,
# None or magnetic variation in degrees (west is negative)
magvar,
)
def _decode_utc(self, utc):
utc_hh, utc_mm, utc_ss = map(float, (utc[:2], utc[2:4], utc[4:]))
return utc_hh * 3600.0 + utc_mm * 60.0 + utc_ss
def _decode_latlon(self, latitude, ns, longitude, ew):
latitude = float(latitude[:2]) + float(latitude[2:])/60.0
if ns == 'S':
latitude = -latitude
longitude = float(longitude[:3]) + float(longitude[3:])/60.0
if ew == 'W':
longitude = -longitude
return (latitude, longitude)
def decode_activesatellites(self, mode1, mode2, *args):
satellites, (pdop, hdop, vdop) = args[:12], map(float, args[12:])
satlist = []
for n in satellites:
if n:
satlist.append(int(n))
else:
satlist.append(None)
mode = (mode1, int(mode2))
return (
# satellite list by channel
tuple(satlist),
# (MODE_AUTO/MODE_FORCED, MODE_NOFIX/MODE_2DFIX/MODE_3DFIX)
mode,
# position dilution of precision
pdop,
# horizontal dilution of precision
hdop,
# vertical dilution of precision
vdop,
)
def decode_fix(self, utc, latitude, ns, longitude, ew, posfix, satellites, hdop, altitude, altitude_units, geoid_separation, geoid_separation_units, dgps_age, dgps_station_id):
latitude, longitude = self._decode_latlon(latitude, ns, longitude, ew)
utc = self._decode_utc(utc)
posfix = int(posfix)
satellites = int(satellites)
hdop = float(hdop)
altitude = (float(altitude), altitude_units)
if geoid_separation != '':
geoid = (float(geoid_separation), geoid_separation_units)
else:
geoid = None
if dgps_age != '':
dgps = (float(dgps_age), dgps_station_id)
else:
dgps = None
return (
# seconds since 00:00 UTC
utc,
# latitude (degrees)
latitude,
# longitude (degrees)
longitude,
# position fix status (POSFIX_INVALID, POSFIX_SPS, POSFIX_DGPS, POSFIX_PPS)
posfix,
# number of satellites used for fix 0 <= satellites <= 12
satellites,
# horizontal dilution of precision
hdop,
# None or (altitude according to WGS-84 ellipsoid, units (typically 'M' for meters))
altitude,
# None or (geoid separation according to WGS-84 ellipsoid, units (typically 'M' for meters))
geoid,
# (age of dgps data in seconds, dgps station id)
dgps,
)
|
