255 lines
8.6 KiB
Python
255 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Analyzer for RTCM message 208 (proprietary).
|
|
Captures message 208 and displays hex dump with field analysis.
|
|
"""
|
|
import base64
|
|
import socket
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
# NTRIP configuration
|
|
CASTER_HOST = "truertk.pointonenav.com"
|
|
CASTER_PORT = 2101
|
|
MOUNTPOINT = "POLARIS_LOCAL"
|
|
USERNAME = "9t7fwfbm57"
|
|
PASSWORD = "96m7bec9g8"
|
|
|
|
LAT = 36.1140884
|
|
LON = -97.0880663
|
|
ALT = 390.0
|
|
|
|
|
|
def build_gga(lat: float, lon: float, alt: float) -> bytes:
|
|
"""Build NMEA GGA sentence."""
|
|
now_utc = datetime.now(timezone.utc).strftime("%H%M%S")
|
|
lat_hemi = "N" if lat >= 0 else "S"
|
|
lat_abs = abs(lat)
|
|
lat_deg = int(lat_abs)
|
|
lat_min = (lat_abs - lat_deg) * 60.0
|
|
lat_str = f"{lat_deg:02d}{lat_min:07.4f}"
|
|
|
|
lon_hemi = "E" if lon >= 0 else "W"
|
|
lon_abs = abs(lon)
|
|
lon_deg = int(lon_abs)
|
|
lon_min = (lon_abs - lon_deg) * 60.0
|
|
lon_str = f"{lon_deg:03d}{lon_min:07.4f}"
|
|
|
|
fields = ["GPGGA", now_utc, lat_str, lat_hemi, lon_str, lon_hemi, "1", "12", "1.0", f"{alt:.1f}", "M", "", "M", "", ""]
|
|
core = ",".join(fields)
|
|
checksum = 0
|
|
for char in core:
|
|
checksum ^= ord(char)
|
|
return f"${core}*{checksum:02X}\r\n".encode("ascii")
|
|
|
|
|
|
def make_ntrip_request(host: str, port: int, mount: str, user: str, password: str) -> bytes:
|
|
"""Create NTRIP v2 HTTP request."""
|
|
auth = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
|
|
req = (
|
|
f"GET /{mount} HTTP/1.1\r\n"
|
|
f"Host: {host}:{port}\r\n"
|
|
f"Ntrip-Version: Ntrip/2.0\r\n"
|
|
f"User-Agent: RTCM-208-Analyzer/1.0\r\n"
|
|
f"Connection: close\r\n"
|
|
f"Authorization: Basic {auth}\r\n\r\n"
|
|
)
|
|
return req.encode("ascii")
|
|
|
|
|
|
def hex_dump(data: bytes, offset: int = 0, width: int = 16) -> str:
|
|
"""Create a hex dump with ASCII representation."""
|
|
lines = []
|
|
for i in range(0, len(data), width):
|
|
chunk = data[i:i+width]
|
|
hex_part = " ".join(f"{b:02X}" for b in chunk)
|
|
ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
|
|
lines.append(f"{offset + i:04X} {hex_part:<{width*3}} {ascii_part}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def analyze_message_208(data: bytes, msg_num: int):
|
|
"""Analyze and display structure of message 208."""
|
|
print(f"\n{'=' * 80}")
|
|
print(f"MESSAGE 208 #{msg_num} - Length: {len(data)} bytes")
|
|
print(f"{'=' * 80}")
|
|
|
|
# Show hex dump
|
|
print("\nHex Dump:")
|
|
print(hex_dump(data))
|
|
|
|
# Try to extract some common fields
|
|
print(f"\n{'─' * 80}")
|
|
print("Field Analysis:")
|
|
print(f"{'─' * 80}")
|
|
|
|
if len(data) >= 3:
|
|
# Message type (12 bits)
|
|
msg_type = (data[0] << 4) | (data[1] >> 4)
|
|
print(f"Message Type: {msg_type}")
|
|
|
|
# Station/Reference ID (typically next 12 bits)
|
|
if len(data) >= 3:
|
|
ref_id = ((data[1] & 0x0F) << 8) | data[2]
|
|
print(f"Reference/Station ID: {ref_id}")
|
|
|
|
# Show first 20 bytes as decimal
|
|
print(f"\nFirst bytes (decimal): {[data[i] for i in range(min(20, len(data)))]}")
|
|
|
|
# Try to find patterns
|
|
print(f"\n{'─' * 80}")
|
|
print("Attempting pattern recognition:")
|
|
print(f"{'─' * 80}")
|
|
|
|
# Look for typical RTCM structures
|
|
if len(data) >= 10:
|
|
# Check for possible timestamp (GPS TOW in ms)
|
|
possible_time_1 = (data[3] << 22) | (data[4] << 14) | (data[5] << 6) | (data[6] >> 2)
|
|
print(f"Possible GPS TOW (ms) at byte 3: {possible_time_1} ({possible_time_1/1000:.1f} sec)")
|
|
|
|
# Check for possible coordinates or large numbers
|
|
for i in range(3, min(len(data) - 4, 10)):
|
|
val32 = (data[i] << 24) | (data[i+1] << 16) | (data[i+2] << 8) | data[i+3]
|
|
if val32 != 0:
|
|
val32_signed = val32 if val32 < 0x80000000 else val32 - 0x100000000
|
|
print(f"32-bit value at byte {i}: {val32} (signed: {val32_signed})")
|
|
|
|
|
|
def main():
|
|
print(f"RTCM Message 208 Analyzer")
|
|
print(f"=" * 80)
|
|
print(f"Caster: {CASTER_HOST}:{CASTER_PORT}/{MOUNTPOINT}")
|
|
print(f"Will capture first 10 instances of message 208")
|
|
print(f"=" * 80)
|
|
print()
|
|
|
|
msg_208_samples = []
|
|
last_gga_time = 0
|
|
start_time = time.monotonic()
|
|
total_messages = 0
|
|
msg_208_count = 0
|
|
|
|
try:
|
|
# Connect
|
|
print("Connecting...")
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(30)
|
|
sock.connect((CASTER_HOST, CASTER_PORT))
|
|
sock.sendall(make_ntrip_request(CASTER_HOST, CASTER_PORT, MOUNTPOINT, USERNAME, PASSWORD))
|
|
|
|
# Read headers
|
|
header = b""
|
|
while b"\r\n\r\n" not in header:
|
|
chunk = sock.recv(1)
|
|
if not chunk:
|
|
print("ERROR: Connection closed")
|
|
return
|
|
header += chunk
|
|
|
|
if "200 OK" not in header.decode("iso-8859-1", errors="replace"):
|
|
print("ERROR: Connection failed")
|
|
return
|
|
|
|
print("Connected! Searching for message 208...\n")
|
|
|
|
# Receive loop
|
|
while msg_208_count < 10:
|
|
now = time.monotonic()
|
|
|
|
# Send GGA every 10 seconds
|
|
if now - last_gga_time >= 10:
|
|
sock.sendall(build_gga(LAT, LON, ALT))
|
|
print(f"[{int(now - start_time):3d}s] Sent GGA | Total msgs: {total_messages}, Msg 208 found: {msg_208_count}")
|
|
last_gga_time = now
|
|
|
|
# Receive data
|
|
data = sock.recv(4096)
|
|
if not data:
|
|
print("Connection closed")
|
|
break
|
|
|
|
# Parse messages
|
|
i = 0
|
|
while i < len(data):
|
|
if data[i] == 0xD3 and i + 2 < len(data):
|
|
length = ((data[i+1] & 0x03) << 8) | data[i+2]
|
|
msg_total_len = 3 + length + 3
|
|
|
|
if i + msg_total_len <= len(data) and length >= 3:
|
|
msg_type = (data[i+3] << 4) | (data[i+4] >> 4)
|
|
msg_data = data[i+3:i+3+length]
|
|
total_messages += 1
|
|
|
|
if msg_type == 208:
|
|
msg_208_count += 1
|
|
msg_208_samples.append(msg_data)
|
|
print(f"✓ Found message 208 #{msg_208_count} (length: {length} bytes)")
|
|
|
|
if msg_208_count <= 3: # Show first 3 immediately
|
|
analyze_message_208(msg_data, msg_208_count)
|
|
|
|
i += msg_total_len
|
|
continue
|
|
i += 1
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\nInterrupted by user")
|
|
except Exception as e:
|
|
print(f"\nERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
if 'sock' in locals():
|
|
try:
|
|
sock.close()
|
|
except:
|
|
pass
|
|
|
|
# Final summary
|
|
print(f"\n{'=' * 80}")
|
|
print(f"SUMMARY")
|
|
print(f"{'=' * 80}")
|
|
print(f"Total messages received: {total_messages}")
|
|
print(f"Message 208 instances: {msg_208_count}")
|
|
|
|
if msg_208_samples:
|
|
# Show statistics
|
|
lengths = [len(m) for m in msg_208_samples]
|
|
print(f"\nMessage 208 length statistics:")
|
|
print(f" Min: {min(lengths)} bytes")
|
|
print(f" Max: {max(lengths)} bytes")
|
|
print(f" Most common: {max(set(lengths), key=lengths.count)} bytes ({lengths.count(max(set(lengths), key=lengths.count))} instances)")
|
|
|
|
# Compare first few messages to find common/changing fields
|
|
if len(msg_208_samples) >= 2:
|
|
print(f"\n{'─' * 80}")
|
|
print("Comparing first 2 messages to identify static vs. dynamic fields:")
|
|
print(f"{'─' * 80}")
|
|
msg1 = msg_208_samples[0]
|
|
msg2 = msg_208_samples[1]
|
|
min_len = min(len(msg1), len(msg2))
|
|
|
|
static_bytes = []
|
|
dynamic_bytes = []
|
|
|
|
for i in range(min_len):
|
|
if msg1[i] == msg2[i]:
|
|
static_bytes.append(i)
|
|
else:
|
|
dynamic_bytes.append(i)
|
|
|
|
print(f"Static byte positions: {static_bytes[:20]}{'...' if len(static_bytes) > 20 else ''}")
|
|
print(f"Dynamic byte positions: {dynamic_bytes[:20]}{'...' if len(dynamic_bytes) > 20 else ''}")
|
|
|
|
print(f"\nByte-by-byte comparison (first 30 bytes):")
|
|
print("Byte# Msg1 Msg2 Same")
|
|
print("─" * 30)
|
|
for i in range(min(30, min_len)):
|
|
same = "✓" if msg1[i] == msg2[i] else "✗"
|
|
print(f"{i:4d} {msg1[i]:02X} {msg2[i]:02X} {same}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|