我使用的是 ZED-F9P。
下面是我编写的 Python 脚本，用于在没有校正数据的情况下打印纬度和经度，但现在我想尝试使用 RTK 获得更准确的结果。
我已经熟悉了应用 RTCM 的桌面应用程序，如 PyGPSClient 和 u-center，但我希望能够在 python 脚本中实现 RTK 修复。
我这么说是因为我的目标是在 Arduino 或类似设备上实现 RTK，然后将其发送到云端，在那里我可以将其与另一个位置的相同设备进行比较（即获取两者之间的距离）。
我想也许我可以使用 PyGPSClient 的部分源代码？我不知道从哪里开始。任何意见，将不胜感激。谢谢！
import serial
gps = serial.Serial('com5', baudrate=9600)
while True:
ser_bytes = gps.readline()
decoded_bytes = ser_bytes.decode("utf-8")
data = decoded_bytes.split(",")
if data[0] == '$GNRMC':
lat_nmea = (data[3],data[4])
lat_degrees = float(lat_nmea[0][0:2])
lat_minutes = float(lat_nmea[0][2:])
lat = lat_degrees + (lat_minutes/60)
lon_nmea = (data[5],data[6])
lon_degrees = float(lon_nmea[0][:3])
lon_minutes = float(lon_nmea[0][3:])
lon = lon_degrees + (lon_minutes/60)
if lat_nmea[1] == 'S':
lat = -lat
if lon_nmea[1] == 'W':
lon = -lon
print("%0.8f" %lat,',' "%0.8f" %lon)
我之前曾尝试通过 pygnssutils 库中的示例 Python 脚本的超链接来回答这个问题，但它被版主删除了。我不确定我是否可以在此处提供完整的脚本（而不是超链接），但假设我可以，这里是：
"""
pygnssutils - rtk_example.py
*** FOR ILLUSTRATION ONLY - NOT FOR PRODUCTION USE ***
PLEASE RESPECT THE TERMS OF USE OF ANY NTRIP CASTER YOU
USE WITH THIS EXAMPLE - INAPPROPRIATE USE CAN RESULT IN
YOUR NTRIP USER ACCOUNT OR IP BEING TEMPORARILY BLOCKED.
This example illustrates how to use the UBXReader and
GNSSNTRIPClient classes to get RTCM3 RTK data from a
designated NTRIP caster/mountpoint and apply it to an
RTK-compatible GNSS receiver (e.g. ZED-F9P) connected to
a local serial port.
GNSSNTRIPClient receives RTK data from the NTRIP caster
and outputs it to a message queue. A send thread reads data
from this queue and sends it to the receiver. A read thread
reads and parses data from the receiver and prints it to the
terminal.
The example also optionally sends NMEA GGA position sentences
to the caster at a prescribed interval, using either fixed
reference coordinates or live coordinates from the receiver.
For brevity, the example will print out just the identities of
all incoming GNSS and NTRIP messages, but the full message can
be printed by setting the global PRINT_FULL variable to True.
The example also includes a simple illustration of horizontal
accuracy. Set the global SHOW_ACCURACY variable to True.
If the receiver is a u-blox UBX receiver, it can be configured
to output UBX RXM-RTCM messages which acknowledge receipt of
incoming RTK data and confirm whether or not it was used (i.e.
RTK correction applied).
NB: Some NTRIP casters may stop sending RTK data after a while
if they're not receiving legitimate NMEA GGA position updates
from the client.
Created on 5 Jun 2022
:author: semuadmin
:copyright: SEMU Consulting © 2022
:license: BSD 3-Clause
"""
# pylint: disable=broad-except
from io import BufferedReader
from threading import Thread, Lock, Event
from queue import Queue
from time import sleep
from serial import Serial
from pyubx2 import (
UBXReader,
NMEA_PROTOCOL,
UBX_PROTOCOL,
RTCM3_PROTOCOL,
protocol,
)
from pyrtcm import RTCM_MSGIDS
from pygnssutils import GNSSNTRIPClient, VERBOSITY_LOW, haversine
# Set to True to print entire GNSS/NTRIP message rather than just identity
PRINT_FULL = False
# Set to True to show estimated horizontal accuracy
SHOW_ACCURACY = True
def read_gnss(stream, lock, stopevent):
"""
THREADED
Reads and parses incoming GNSS data from receiver.
"""
ubr = UBXReader(
BufferedReader(stream),
protfilter=(NMEA_PROTOCOL | UBX_PROTOCOL | RTCM3_PROTOCOL),
)
while not stopevent.is_set():
try:
if stream.in_waiting:
lock.acquire()
(raw_data, parsed_data) = ubr.read() # pylint: disable=unused-variable
lock.release()
if parsed_data:
idy = parsed_data.identity
if SHOW_ACCURACY:
# show estimated horizontal accuracy and distance between receiver
# coordinates and fixed reference point
if hasattr(parsed_data, "lat") and hasattr(parsed_data, "lon"):
lat = parsed_data.lat
lon = parsed_data.lon
dev = haversine(lat, lon, REFLAT, REFLON) * 1000 # meters
print(
f"Receiver coordinates: {lat}, {lon}\r\n",
f"Approximate deviation from fixed ref: {dev:06,f} m",
)
if hasattr(parsed_data, "hAcc"):
unit = 1 if idy == "PUBX" else 0.001
print(
f"Estimated horizontal accuracy: {(parsed_data.hAcc * unit):.3f} m"
)
# if it's an RXM-RTCM message, show which RTCM3 message
# it's acknowledging and whether it's been used or not.""
if idy == "RXM-RTCM":
nty = (
f" - {parsed_data.msgType} "
f"{'Used' if parsed_data.msgUsed > 0 else 'Not used'}"
)
else:
nty = ""
if PRINT_FULL:
print(parsed_data)
else:
print(f"GNSS>> {idy}{nty}")
except Exception as err:
print(f"Something went wrong in read thread {err}")
break
def send_gnss(stream, lock, stopevent, inqueue):
"""
THREADED
Reads RTCM3 data from message queue and sends it to receiver.
"""
while not stopevent.is_set():
try:
raw_data, parsed_data = inqueue.get()
if protocol(raw_data) == RTCM3_PROTOCOL:
if PRINT_FULL:
print(parsed_data)
else:
print(
f"NTRIP>> {parsed_data.identity} {RTCM_MSGIDS[parsed_data.identity]}"
)
lock.acquire()
stream.write(raw_data)
lock.release()
except Exception as err:
print(f"Something went wrong in send thread {err}")
break
if __name__ == "__main__":
# pylint: disable=invalid-name
# GNSS receiver serial port parameters - AMEND AS REQUIRED:
SERIAL_PORT = "/dev/tty.usbmodem1301"
BAUDRATE = 38400
TIMEOUT = 0.1
# NTRIP caster parameters - AMEND AS REQUIRED:
# Ideally, mountpoint should be <30 km from location.
NTRIP_SERVER = "rtk2go.com"
NTRIP_PORT = 2101
MOUNTPOINT = "WEBBPARTNERS"
NTRIP_USER = "[email protected]"
NTRIP_PASSWORD = "password"
# NMEA GGA sentence status - AMEND AS REQUIRED:
GGAMODE = 1 # use fixed reference position (0 = use live position)
GGAINT = 10 # interval in seconds (-1 = do not send NMEA GGA sentences)
# Fixed reference coordinates (used when GGAMODE = 1) - AMEND AS REQUIRED:
REFLAT = 53
REFLON = -2.4
REFALT = 40
REFSEP = 0
serial_lock = Lock()
ntrip_queue = Queue()
stop = Event()
try:
print(f"Opening serial port {SERIAL_PORT} @ {BAUDRATE}...\n")
with Serial(SERIAL_PORT, BAUDRATE, timeout=TIMEOUT) as serial:
stop.clear()
print("Starting read thread...\n")
read_thread = Thread(
target=read_gnss,
args=(
serial,
serial_lock,
stop,
),
daemon=True,
)
read_thread.start()
print("Starting send thread...\n")
send_thread = Thread(
target=send_gnss,
args=(
serial,
serial_lock,
stop,
ntrip_queue,
),
daemon=True,
)
send_thread.start()
print(f"Starting NTRIP client on {NTRIP_SERVER}:{NTRIP_PORT}...\n")
with GNSSNTRIPClient(None, verbosity=VERBOSITY_LOW) as gnc:
streaming = gnc.run(
server=NTRIP_SERVER,
port=NTRIP_PORT,
mountpoint=MOUNTPOINT,
user=NTRIP_USER,
password=NTRIP_PASSWORD,
reflat=REFLAT,
reflon=REFLON,
refalt=REFALT,
refsep=REFSEP,
ggamode=GGAMODE,
ggainterval=GGAINT,
output=ntrip_queue,
)
while streaming and not stop.is_set(): # run until user presses CTRL-C
sleep(1)
sleep(1)
except KeyboardInterrupt:
stop.set()
print("Terminated by user")