- refactored
- added claude.md
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
# CLAUDE.md
|
||||
|
||||
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||||
|
||||
## What this project is
|
||||
|
||||
A Python client for u-blox GNSS receivers (NEO-F9P, ZED-X20P) connected over TCP/IP. It receives and parses both NMEA 0183 and UBX binary protocol messages, and can send UBX poll requests to the receivers.
|
||||
|
||||
Hardware reference documentation is in `readme.md` (u-blox datasheets, integration manuals, interface specs, RTK/RTCM references).
|
||||
|
||||
## Running
|
||||
|
||||
```bash
|
||||
python network_backend.py
|
||||
```
|
||||
|
||||
The entry point in `network_backend.py` connects to two GNSS receivers, wires up packetizers and message listeners, then polls UBX messages every second.
|
||||
|
||||
## Architecture
|
||||
|
||||
The design follows a **listener/talker pipeline** pattern:
|
||||
|
||||
```
|
||||
NetworkBackend (TCP socket per receiver)
|
||||
└─> Transceiver (one per receiver, named to match socket)
|
||||
├─> NmeaPacketizer ──> NmeaListener subclasses (Gga, Gll, Gsa, Gsv, Rmc)
|
||||
└─> UbxPacketizer ──> UbxListener instances (wrapping UbxMsg subclasses)
|
||||
```
|
||||
|
||||
**Core abstractions** (`msg/`):
|
||||
- `MsgContainer` — wraps raw `bytes` with a `source` (ATransceiver) and timestamp
|
||||
- `MsgListener` — ABC with `on_recv(msg)` and `is_msg(msg)` (default: False); only called when `is_msg` returns True
|
||||
- `MsgTalker` — holds a list of `MsgListener` sinks; `call_listener` iterates them and dispatches
|
||||
|
||||
**Transceiver** (`transceiver.py`): inherits both `ATransceiver` (MsgListener) and `MsgTalker`. Receives raw bytes from the backend via `on_recv`, wraps them in `MsgContainer`, and fans out to downstream listeners. Sends bytes back via `backend.send`.
|
||||
|
||||
**NetworkBackend** (`network_backend.py`): manages non-blocking TCP sockets via `selectors`. Each connection has a name, address, associated `Transceiver`, and outgoing `Queue`. Call `register_xcvr(xcvr)` to bind a transceiver to its socket by name match, then `connect()` + `start()` to begin the event loop thread.
|
||||
|
||||
**NMEA parsing** (`nmea/`):
|
||||
- `NmeaPacketizer`: buffers bytes and emits a `MsgContainer` per line (CR/LF delimited)
|
||||
- `NmeaListener` / `AsciiListener`: filters by regex on the raw bytes; subclasses override `on_recv` to parse CSV fields
|
||||
- Message classes (`nmea/messages/`): `Gga`, `Gll`, `Gsa`, `Gsv`, `Rmc` — each takes `(name, tid)` where `tid` is the talker ID (e.g. `'GN'`, `'GP'`); use `'--'` to match any talker
|
||||
|
||||
**UBX parsing** (`ubx/`):
|
||||
- `ubx.py`: `frame_create(class, id, data)` builds a framed UBX packet with Fletcher checksum; `frame_parse` / `frame_parse_hdr` decode received frames
|
||||
- `UbxPacketizer`: buffers bytes, detects sync word `0xB5 0x62`, reassembles complete frames by checking the length field and verifying checksum
|
||||
- `UbxMsg` (base class in `ubx/messages/msg.py`): holds `mclass` / `mid`; subclasses implement `from_bytes(data)` as a classmethod
|
||||
- `UbxListener`: wraps a `UbxMsg`, matches by class+id in `is_msg`, parses and prints in `on_recv`
|
||||
- `UbxQuery` (`ubx/query.py`): extends `UbxListener` with a semaphore-based synchronous `poll()` — sends a UBX poll frame and blocks until the response arrives
|
||||
|
||||
**Implemented UBX messages** (`ubx/messages/`):
|
||||
- `nav.py`: `Sig` (0x01/0x43 NAV-SIG), `Sat` (0x01/0x35 NAV-SAT)
|
||||
- `rxm.py`: `RawX` (0x02/0x15 RXM-RAWX)
|
||||
|
||||
## Adding a new UBX message
|
||||
|
||||
1. Create a subclass of `UbxMsg` with the correct `mclass`/`mid` and a `from_bytes` classmethod
|
||||
2. Instantiate `UbxListener(YourMsg())` and register it on the `UbxPacketizer`
|
||||
|
||||
## Adding a new NMEA sentence
|
||||
|
||||
Subclass `NmeaListener` with the appropriate `msg_type` string, override `on_recv` to split on `b','` and populate a `Msg` dataclass, then register on the `NmeaPacketizer`.
|
||||
+7
-6
@@ -14,6 +14,7 @@ from nmea.packetizer import NmeaPacketizer
|
||||
from nmea.messages import Gll, Gsa, Gsv, Gga, Rmc
|
||||
|
||||
from ubx.ubx import frame_create
|
||||
from ubx.msg_types import *
|
||||
from ubx.packetizer import UbxPacketizer
|
||||
from ubx.messages import Sig, Sat, RawX, UbxMsg
|
||||
from ubx.listener import UbxListener
|
||||
@@ -146,8 +147,8 @@ if __name__ == "__main__":
|
||||
UbxListener(Sat()),
|
||||
UbxListener(RawX()),
|
||||
|
||||
UbxListener(UbxMsg(0x05, 0x01)),
|
||||
UbxListener(UbxMsg(0x05, 0x00)),
|
||||
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_ACK)),
|
||||
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_NACK)),
|
||||
]
|
||||
|
||||
# Topology
|
||||
@@ -192,16 +193,16 @@ if __name__ == "__main__":
|
||||
try:
|
||||
for grm in [" neo-f9p", "zed-x20p"]:
|
||||
# Poll UBX-NAV-CLOCK (0x01 0x22)
|
||||
frame = frame_create(0x01, 0x22, b'')
|
||||
frame = frame_create(UBX_CLASS_NAV, UBX_ID_CLOCK)
|
||||
transceiver[grm].send(frame)
|
||||
# Poll UBX-NAV-SIG (0x01 0x43)
|
||||
frame = frame_create(0x01, 0x43, b'')
|
||||
frame = frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SIG)
|
||||
transceiver[grm].send(frame)
|
||||
# Poll UBX-NAV-SAT (0x01 0x35)
|
||||
frame = frame_create(0x01, 0x35, b'')
|
||||
frame = frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SAT)
|
||||
transceiver[grm].send(frame)
|
||||
# Poll UBX-RXM-RAWX (0x02 0x15)
|
||||
frame = frame_create(0x02, 0x15, b'')
|
||||
frame = frame_create(UBX_CLASS_RXM, UBX_ID_RXM_RAW)
|
||||
transceiver[grm].send(frame)
|
||||
except ValueError:
|
||||
break
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
UBX_SYNC1 = 0xB5
|
||||
UBX_SYNC2 = 0x62
|
||||
|
||||
UBX_CLASS_NAV = 0x01
|
||||
UBX_ID_CLOCK = 0x22
|
||||
UBX_ID_NAV_SVINFO = 0x30
|
||||
UBX_ID_NAV_SAT = 0x35
|
||||
UBX_ID_NAV_SIG = 0x43
|
||||
|
||||
UBX_CLASS_RXM = 0x02
|
||||
UBX_ID_RXM_RAW = 0x15
|
||||
|
||||
UBX_CLASS_ACK = 0x05
|
||||
UBX_ID_ACK_NACK = 0x00
|
||||
UBX_ID_ACK_ACK = 0x01
|
||||
+5
-4
@@ -1,6 +1,8 @@
|
||||
import struct
|
||||
|
||||
UBX_SYNC_WORD = b'\xb5\x62'
|
||||
from ubx.msg_types import *
|
||||
|
||||
UBX_SYNC_WORD = struct.pack('BB', UBX_SYNC1, UBX_SYNC2)
|
||||
|
||||
def checksum(data: bytes) -> bytes:
|
||||
ck_a = 0
|
||||
@@ -11,14 +13,13 @@ def checksum(data: bytes) -> bytes:
|
||||
|
||||
return struct.pack("BB", ck_a, ck_b)
|
||||
|
||||
def frame_create(_class: int, _id: int, _data: bytes) -> bytes:
|
||||
_sync = b'\xb5\x62'
|
||||
def frame_create(_class: int, _id: int, _data: bytes = b'') -> bytes:
|
||||
_hdr = struct.pack('B', _class)
|
||||
_hdr += struct.pack('B', _id)
|
||||
_hdr += struct.pack('<H', len(_data))
|
||||
_check = checksum(_hdr + _data)
|
||||
|
||||
return _sync + _hdr + _data + _check
|
||||
return UBX_SYNC_WORD + _hdr + _data + _check
|
||||
|
||||
class Hdr:
|
||||
def __init__(self, _class: int, _id: int, _length: int):
|
||||
|
||||
Reference in New Issue
Block a user