jt808-python-parser

from parser import parse_jt808_frame, decode_alarm_bits, decode_status_bits

# Raw JT808 frame bytes (must start and end with 0x7E)
frame = bytes.fromhex("7e020000...")

pkt = parse_jt808_frame(frame)

print(f"Message ID : 0x{pkt.msg_id:04X}")
print(f"Phone      : {pkt.header.phone}")
print(f"Flow ID    : {pkt.header.flow_id}")

if pkt.location:
    loc = pkt.location
    print(f"Latitude   : {loc.latitude}")
    print(f"Longitude  : {loc.longitude}")
    print(f"Altitude   : {loc.altitude} m")
    print(f"Speed      : {loc.speed} km/h")
    print(f"Direction  : {loc.direction}°")
    print(f"Time       : {loc.time}")          # YYMMDDhhmmss
    print(f"Alarms     : {decode_alarm_bits(loc.alarm)}")
    print(f"Status     : {decode_status_bits(loc.status)}")
    print(f"Extra items: {loc.extra}")

import struct
from parser import build_jt808_frame, MSG_LOCATION_REPORT

alarm     = 0
status    = 0b11          # ACC on + positioned
lat_raw   = int(39.774 * 1_000_000)
lon_raw   = int(116.352 * 1_000_000)
altitude  = 50            # metres
speed_raw = 600           # 60.0 km/h (unit: 1/10 km/h)
direction = 90            # degrees
time_bcd  = bytes([0x24, 0x01, 0x15, 0x12, 0x30, 0x00])  # 2024-01-15 12:30:00

body = struct.pack(">IIIIHHH", alarm, status, lat_raw, lon_raw, altitude, speed_raw, direction)
body += time_bcd

frame = build_jt808_frame(
    msg_id=MSG_LOCATION_REPORT,
    body=body,
    phone="013800001234",
    flow_id=1,
)
# Send `frame` over your TCP/UDP connection

from parser import parse_jt808_frame, MSG_TERMINAL_REGISTER

pkt = parse_jt808_frame(frame_bytes)
if pkt.register:
    reg = pkt.register
    print(f"Province ID        : {reg.province_id}")
    print(f"City ID            : {reg.city_id}")
    print(f"Manufacturer ID    : {reg.manufacturer_id}")
    print(f"Terminal Model     : {reg.terminal_model}")
    print(f"Terminal ID        : {reg.terminal_id}")
    print(f"License Plate Color: {reg.license_plate_color}")
    print(f"License Plate      : {reg.license_plate}")

parse_jt808_frame(frame: bytes) -> JT808Packet

parse_header(data: bytes) -> tuple[JT808Header, int]

parse_0200_location(body: bytes) -> JT808Location

parse_0100_register(body: bytes) -> JT808Register

parse_0102_auth(body: bytes) -> JT808Auth

parse_8001_platform_response(body: bytes) -> JT808PlatformResponse

build_jt808_frame(msg_id, body, phone, flow_id, encryption=0,
                  has_subpackage=False, total_subpackages=0, subpackage_seq=0) -> bytes

build_header(msg_id, body, phone, flow_id, ...) -> bytes

unescape(data: bytes) -> bytes      # Remove JT808 byte stuffing
escape(data: bytes) -> bytes        # Apply JT808 byte stuffing
calc_checksum(data: bytes) -> int   # XOR checksum over all bytes
bcd_to_str(b: bytes) -> str        # Decode packed BCD to digit string
hexstr_to_bytes(s: str) -> bytes   # Parse hex string to bytes
bytes_to_hex(b: bytes) -> str      # Format bytes as hex string

decode_alarm_bits(alarm: int) -> list[str]   # Active alarm descriptions
decode_status_bits(status: int) -> list[str] # Active status descriptions