Skip to content

README.md

Latest commit

 

History

History
102 lines (70 loc) · 2.42 KB

README.md

File metadata and controls

102 lines (70 loc) · 2.42 KB
Raw

NTRIP Client

A lightweight NTRIP client for connecting to NTRIP casters and streaming RTCM correction data.

Usage

from ntrip import NTRIPClient

client = NTRIPClient("caster.example.com", 2101, "MOUNTPOINT")
client.connect(user="user", password="pass")

# Send NMEA GGA position
client.write(gga.encode())
# Read RTCM correction data
data = client.read()

client.close()

Authentication

Supports no-auth, HTTP Basic, and Bearer token authentication:

client.connect()
client.connect(user="user", password="pass")
client.connect(token="my-token")

Options

connect() also accepts:

  • timeout — handshake timeout in seconds (default: 5)
  • user_agent — User-Agent header (default: "NTRIP Ardusimple")
  • reconnect — number of automatic reconnection attempts (default: 0)
client.connect(user="user", password="pass", timeout=15, user_agent="MyApp/1.0", reconnect=3)

Context manager

Ensures the connection is closed on exit, even if an exception occurs:

with NTRIPClient("caster.example.com", 2101, "MOUNTPOINT") as client:
    client.connect(user="user", password="pass")
    while True:
        data = client.read()

Reconnection

If reconnect is set, failed read() or write() calls automatically reconnect with linear backoff between attempts. On success, the retry counter resets:

client.connect(user="user", password="pass", reconnect=3)

# Will attempt up to 3 reconnections with 1s, 2s delays
data = client.read()

Async GPS Bridge

NTRIPBridge connects your GPS device to the NTRIP correction service, sending your position and forwarding corrections back asynchronously. Wrap client in AsyncClientAdapter to use it with the bridge.

from asyncio import run, create_task, sleep
from ntrip import NTRIPClient, AsyncClientAdapter, NTRIPBridge

async def main():
    reader, writer = ...  #  StreamReader/StreamWriter pair

    client = AsyncClientAdapter(NTRIPClient("caster.example.com", 2101, "MOUNTPOINT"))
    await client.connect(user="user", password="pass", reconnect=3)

    bridge = NTRIPBridge(client, reader, writer)
    create_task(bridge.start()) # Runs in the background

    while True:
        # Main async loop...
        await sleep(1)

if __name__ == '__main__':
    run(main())

Development

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Run the tests:

pytest -v