app.ndtp.protocol

NDTP wire codec: stream framing, CRC, and decoding of what the telematics terminals send.

Pure functions and dataclasses only — no sockets, no asyncio, no app state — so the codec is unit-testable on raw bytes and reusable by the TCP server, replay tools and fake-device clients.

Wire format (all integers little-endian, packed). Verified against the emulator’s own bytecode (ndtp-telemetry-emulator:1.0, CellFactory.wrapNphPacket), not just the spec:

frame = NPL header (15 B) | NPH header (10 B) | body
NPL   = signature u16 (0x7E7E) | data_size u16 (= len(NPH + body)) | flags u16
      | crc u16 | type u8 (2 = NPH) | peer_address u32 (= unitId) | request_id u16
NPH   = service_id u16 | type u16 | flags u16 (bit0 = request) | request_id u32

CRC is CRC-16/MODBUS over NPH + body. The emulator byte-swaps it before writing it into the little-endian field, so on the wire it reads as big-endian. The NPL crc flag bit is always 0, yet the CRC is always filled in — so we always verify it.

Realtime bodies are a sequence of cells [type u8][number u8][payload] with no length prefix: the payload size is implied by the type. A cell type we don’t know the size of therefore ends parsing of that packet (see Realtime.unparsed_tail). G6CellNav00 is always first, so navigation is never lost to this.

Functions

`crc16_modbus`(data)	CRC-16/MODBUS: init 0xFFFF, reflected poly 0xA001, no final xor.
`decode_frame`(frame)	Interpret a frame's NPH header and body.
`encode_frame`(service_id, nph_type, body, *, ...)
`encode_handshake`(unit_id, *[, request_id, ...])
`encode_nav_cell`(*, timestamp, latitude, ...)	A complete `G6CellNav00` cell (2-byte cell header + 26-byte payload).
`encode_realtime`(unit_id, cells, *, request_id)

Classes

`DecoderStats`([frames, bytes_discarded, ...])
`Frame`(npl, payload)	One CRC-verified NPL frame.
`FrameDecoder`(*[, max_data_size])	Incremental decoder for one TCP connection: `feed()` raw bytes, get complete frames.
`Handshake`(nph, proto_version, flags, ...)	`NPH_SGC_CONN_REQUEST` — the first packet on every (re)connection.
`NavCell`(number, timestamp, latitude, ...)	`G6CellNav00`: one navigation fix.
`NphHeader`(service_id, type, flags, request_id)
`NplHeader`(data_size, flags, crc, type, ...)
`RawCell`(type, number, payload)	A cell of known size whose fields we don't decode.
`Realtime`(nph, nav, cells[, unparsed_tail])	`NPH_SND_REALTIME` — a telemetry packet.
`UnknownMessage`(nph, body)

Exceptions

NdtpDecodeError

A CRC-valid frame whose content doesn't match its declared message type.

class app.ndtp.protocol.DecoderStats(frames: 'int' = 0, bytes_discarded: 'int' = 0, crc_errors: 'int' = 0, bad_headers: 'int' = 0, unsupported_type: 'int' = 0)[исходный код]

Базовые классы: object

__init__(frames: int = 0, bytes_discarded: int = 0, crc_errors: int = 0, bad_headers: int = 0, unsupported_type: int = 0) → None

bad_headers: int

bytes_discarded: int

crc_errors: int

frames: int

unsupported_type: int

class app.ndtp.protocol.Frame(npl: NplHeader, payload: bytes)[исходный код]

Базовые классы: object

One CRC-verified NPL frame. payload is the NPH header + body.

__init__(npl: NplHeader, payload: bytes) → None

npl: NplHeader

payload: bytes

class app.ndtp.protocol.FrameDecoder(*, max_data_size: int = 4096)[исходный код]

Базовые классы: object

Incremental decoder for one TCP connection: feed() raw bytes, get complete frames.

TCP is a byte stream, so a read may hold half a frame or several frames. Framing errors never raise: garbage is skipped, a bad frame is dropped and the decoder resyncs on the next signature, with everything counted in stats.

__init__(*, max_data_size: int = 4096) → None[исходный код]

feed(data: bytes | bytearray | memoryview) → list[Frame][исходный код]

class app.ndtp.protocol.Handshake(nph: NphHeader, proto_version: tuple[int, int], flags: int, peer_address: int, max_packet_size: int)[исходный код]

Базовые классы: object

NPH_SGC_CONN_REQUEST — the first packet on every (re)connection.

__init__(nph: NphHeader, proto_version: tuple[int, int], flags: int, peer_address: int, max_packet_size: int) → None

flags: int

max_packet_size: int

nph: NphHeader

peer_address: int

proto_version: tuple[int, int]

class app.ndtp.protocol.NavCell(number: int, timestamp: int, latitude: float, longitude: float, location_valid: bool, flags: int, bat_voltage_mv: int, speed_avg_kmh: int, speed_max_kmh: int, course_deg: int, track_m: int, altitude_m: int, nsat: int, pdop: int)[исходный код]

Базовые классы: object

G6CellNav00: one navigation fix.

__init__(number: int, timestamp: int, latitude: float, longitude: float, location_valid: bool, flags: int, bat_voltage_mv: int, speed_avg_kmh: int, speed_max_kmh: int, course_deg: int, track_m: int, altitude_m: int, nsat: int, pdop: int) → None

property alarm: bool

altitude_m: int

bat_voltage_mv: int

course_deg: int

flags: int

latitude: float

location_valid: bool

longitude: float

nsat: int

number: int

pdop: int

property sos: bool

speed_avg_kmh: int

speed_max_kmh: int

timestamp: int

track_m: int

exception app.ndtp.protocol.NdtpDecodeError[исходный код]

Базовые классы: ValueError

A CRC-valid frame whose content doesn’t match its declared message type.

class app.ndtp.protocol.NphHeader(service_id: 'int', type: 'int', flags: 'int', request_id: 'int')[исходный код]

Базовые классы: object

__init__(service_id: int, type: int, flags: int, request_id: int) → None

flags: int

property is_request: bool

request_id: int

service_id: int

type: int

class app.ndtp.protocol.NplHeader(data_size: 'int', flags: 'int', crc: 'int', type: 'int', peer_address: 'int', request_id: 'int')[исходный код]

Базовые классы: object

__init__(data_size: int, flags: int, crc: int, type: int, peer_address: int, request_id: int) → None

crc: int

data_size: int

flags: int

peer_address: int

request_id: int

type: int

class app.ndtp.protocol.RawCell(type: int, number: int, payload: bytes)[исходный код]

Базовые классы: object

A cell of known size whose fields we don’t decode.

__init__(type: int, number: int, payload: bytes) → None

number: int

payload: bytes

type: int

class app.ndtp.protocol.Realtime(nph: NphHeader, nav: NavCell | None, cells: tuple[NavCell | RawCell, ...], unparsed_tail: bytes = b'')[исходный код]

Базовые классы: object

NPH_SND_REALTIME — a telemetry packet.

__init__(nph: NphHeader, nav: NavCell | None, cells: tuple[NavCell | RawCell, ...], unparsed_tail: bytes = b'') → None

cells: tuple[NavCell | RawCell, ...]

nav: NavCell | None

nph: NphHeader

unparsed_tail: bytes

class app.ndtp.protocol.UnknownMessage(nph: 'NphHeader', body: 'bytes')[исходный код]

Базовые классы: object

__init__(nph: NphHeader, body: bytes) → None

body: bytes

nph: NphHeader

app.ndtp.protocol.crc16_modbus(data: bytes | bytearray | memoryview) → int[исходный код]: CRC-16/MODBUS: init 0xFFFF, reflected poly 0xA001, no final xor.

app.ndtp.protocol.decode_frame(frame: Frame) → Handshake | Realtime | UnknownMessage[исходный код]: Interpret a frame’s NPH header and body. Raises NdtpDecodeError on malformed content.

app.ndtp.protocol.encode_frame(service_id: int, nph_type: int, body: bytes, *, peer_address: int, nph_request_id: int = 0, request: bool = True) → bytes[исходный код]

app.ndtp.protocol.encode_handshake(unit_id: int, *, request_id: int = 1, max_packet_size: int = 65535) → bytes[исходный код]

app.ndtp.protocol.encode_nav_cell(*, timestamp: int, latitude: float, longitude: float, valid: bool = True, speed_kmh: int = 0, course_deg: int = 0, altitude_m: int = 0, number: int = 0) → bytes[исходный код]: A complete G6CellNav00 cell (2-byte cell header + 26-byte payload).

app.ndtp.protocol.encode_realtime(unit_id: int, cells: bytes, *, request_id: int) → bytes[исходный код]