usr2gpx.py

#!/usr/bin/env python3
"""Convert Navico/B&G/Lowrance/Simrad .usr files (v4/5/6) to GPX.

Extracts tracks (trails) only. Based on the GPSBabel lowranceusr.cc reference.

Note: completely vibe-coded, but works for my .USR files
"""

import struct
import math
import sys
import os
from datetime import datetime, timezone, timedelta


DEGREESTORADIANS = math.pi / 180.0


class USRReader:
    def __init__(self, data: bytes):
        self.data = data
        self.pos = 0

    def read_int16(self) -> int:
        v = struct.unpack_from('<h', self.data, self.pos)[0]
        self.pos += 2
        return v

    def read_uint16(self) -> int:
        v = struct.unpack_from('<H', self.data, self.pos)[0]
        self.pos += 2
        return v

    def read_int32(self) -> int:
        v = struct.unpack_from('<i', self.data, self.pos)[0]
        self.pos += 4
        return v

    def read_uint32(self) -> int:
        v = struct.unpack_from('<I', self.data, self.pos)[0]
        self.pos += 4
        return v

    def read_float(self) -> float:
        v = struct.unpack_from('<f', self.data, self.pos)[0]
        self.pos += 4
        return v

    def read_double(self) -> float:
        v = struct.unpack_from('<d', self.data, self.pos)[0]
        self.pos += 8
        return v

    def read_byte(self) -> int:
        v = self.data[self.pos]
        self.pos += 1
        return v

    def read_bytes(self, n: int) -> bytes:
        v = self.data[self.pos:self.pos + n]
        self.pos += n
        return v

    def read_str(self, bytes_per_char: int) -> str:
        """Read a length-prefixed string (matches lowranceusr4_readstr)."""
        length = self.read_int32()  # length in bytes
        if length <= 0:
            return ""
        raw = self.read_bytes(length)
        if bytes_per_char == 1:
            return raw.decode('utf-8', errors='replace')
        else:
            return raw.decode('utf-16-le', errors='replace')

    def eof(self) -> bool:
        return self.pos >= len(self.data)


def jdn_ms_to_datetime(jdn: int, ms: int) -> datetime:
    """Convert Julian Day Number + milliseconds to datetime."""
    # JDN 2440588 = 1970-01-01
    days_since_epoch = jdn - 2440588
    seconds = days_since_epoch * 86400 + ms // 1000
    return datetime.fromtimestamp(seconds, tz=timezone.utc)


def parse_usr(data: bytes):
    """Parse a USR file and return list of trails with their track points."""
    r = USRReader(data)

    # File header
    version = r.read_uint16()
    rstream = r.read_uint16()

    if version < 4 or version > 6:
        raise ValueError(f"USR version {version} not supported (only v4/5/6)")

    # v4+ header fields
    unknown = r.read_int32()
    title = r.read_str(1)
    date_str = r.read_str(1)
    create_jdn = r.read_uint32()
    create_time = r.read_uint32()
    _unused = r.read_byte()
    serial = r.read_uint32()
    comment = r.read_str(1)

    print(f"USR version {version}, stream {rstream}")
    print(f"Title: {title}")
    print(f"Date: {date_str}")
    print(f"Serial: {serial}")
    print(f"Comment: {comment}")

    # --- Skip waypoints ---
    num_waypoints = r.read_int32()
    print(f"Waypoints: {num_waypoints} (skipping)")

    for i in range(num_waypoints):
        if r.eof():
            break
        _skip_waypoint_v456(r, version)

    # --- Skip routes ---
    num_routes = r.read_int32()
    print(f"Routes: {num_routes} (skipping)")

    for i in range(num_routes):
        if r.eof():
            break
        _skip_route_v456(r, version)

    # --- Parse trails ---
    num_trails = r.read_int32()
    print(f"Trails: {num_trails}")

    trails = []
    for i in range(num_trails):
        if r.eof():
            break
        trail = _parse_trail_v456(r, version)
        trails.append(trail)
        print(f"  Trail '{trail['name']}': {len(trail['points'])} points")

    return trails


def _skip_waypoint_v456(r: USRReader, version: int):
    """Skip a single waypoint record for USR v4/5/6."""
    if version > 4:
        r.read_bytes(16)  # UUID (4 x int32)

    r.read_uint32()   # uid_unit
    r.read_int32()    # uid_seq_low
    r.read_int32()    # uid_seq_high
    r.read_uint16()   # stream version

    r.read_str(2)     # name (UTF-16)

    if version > 4:
        r.read_uint32()  # uid_unit2

    r.read_int32()    # longitude (mercator)
    r.read_int32()    # latitude (mercator)
    r.read_int32()    # flags
    r.read_uint16()   # icon id
    r.read_uint16()   # color id
    r.read_str(2)     # description (UTF-16)
    r.read_float()    # alarm radius
    r.read_uint32()   # creation date JDN
    r.read_uint32()   # creation time ms
    r.read_byte()     # unused
    r.read_float()    # depth
    r.read_int32()    # loran GRI
    r.read_int32()    # loran Tda
    r.read_int32()    # loran Tdb


def _skip_route_v456(r: USRReader, version: int):
    """Skip a single route record for USR v4/5/6."""
    if version >= 5:
        r.read_bytes(16)  # UUID

    r.read_uint32()   # uid_unit
    r.read_int32()    # uid_seq_low
    r.read_int32()    # uid_seq_high
    r.read_uint16()   # route stream version

    r.read_str(2)     # name (UTF-16)

    if version >= 5:
        r.read_uint32()  # unit id

    num_legs = r.read_int32()

    if version <= 4:
        # UID-based: 3 x int32 per leg
        for _ in range(num_legs):
            r.read_bytes(12)
    else:
        # UUID-based: 4 x int32 per leg
        for _ in range(num_legs):
            r.read_bytes(16)

    if version > 4:
        r.read_int32()   # mystery
        r.read_int32()   # mystery
        r.read_byte()    # mystery

    r.read_byte()  # mystery end byte


def _parse_trail_v456(r: USRReader, version: int) -> dict:
    """Parse a single trail record for USR v4/5/6."""
    # UID
    uid_unit = r.read_uint32()
    uid_seq_low = r.read_int32()
    uid_seq_high = r.read_int32()

    # Trail stream version
    trail_version = r.read_uint16()

    # Name (UTF-16)
    name = r.read_str(2)

    # Flags and color
    flags = r.read_int32()
    color = r.read_int32()

    # Description (UTF-16)
    desc = r.read_str(2)

    # Creation date/time
    create_date = r.read_uint32()
    create_time = r.read_uint32()

    # 3 flag bytes
    r.read_byte()
    r.read_byte()
    r.read_byte()

    # Attribute data (trail version 6 has no header attributes)
    if trail_version < 6:
        attr_count = r.read_int32()
        for _ in range(attr_count):
            if trail_version == 5:
                r.read_int32()
            else:
                r.read_byte()

    # Trail points
    num_points = r.read_int32()
    points = []

    for _ in range(num_points):
        # Unknown header bytes
        r.read_uint16()
        r.read_byte()

        # POSIX timestamp
        timestamp = r.read_uint32()

        # Longitude and latitude in radians (IEEE 754 double)
        lon_rad = r.read_double()
        lat_rad = r.read_double()
        lon_deg = lon_rad / DEGREESTORADIANS
        lat_deg = lat_rad / DEGREESTORADIANS

        # Per-point attributes
        m = r.read_int32()
        attrs = {}
        for _ in range(m):
            flag = r.read_byte()
            value = r.read_float()
            attrs[flag] = value

        dt = datetime.fromtimestamp(timestamp, tz=timezone.utc) if timestamp > 0 else None
        points.append({
            'lat': lat_deg,
            'lon': lon_deg,
            'time': dt,
            'attrs': attrs,
        })

    return {
        'name': name,
        'desc': desc,
        'trail_version': trail_version,
        'points': points,
    }


def trails_to_gpx(trails: list, creator: str = "usr2gpx") -> str:
    """Convert parsed trails to GPX XML string."""
    lines = [
        '<?xml version="1.0" encoding="UTF-8"?>',
        '<gpx version="1.1" creator="{}"'.format(creator),
        '     xmlns="http://www.topografix.com/GPX/1/1"',
        '     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"',
        '     xsi:schemaLocation="http://www.topografix.com/GPX/1/1'
        ' http://www.topografix.com/GPX/1/1/gpx.xsd">',
    ]

    for trail in trails:
        lines.append('  <trk>')
        if trail['name']:
            lines.append(f'    <name>{_xml_escape(trail["name"])}</name>')
        if trail['desc']:
            lines.append(f'    <desc>{_xml_escape(trail["desc"])}</desc>')
        lines.append('    <trkseg>')
        for pt in trail['points']:
            time_attr = ''
            if pt['time']:
                time_attr = f'      <time>{pt["time"].strftime("%Y-%m-%dT%H:%M:%SZ")}</time>'

            # Extract speed if present (flag 0x01)
            speed_elem = ''
            if 0x01 in pt['attrs']:
                speed_elem = f'      <speed>{pt["attrs"][0x01]:.2f}</speed>'

            lines.append(f'    <trkpt lat="{pt["lat"]:.9f}" lon="{pt["lon"]:.9f}">')
            if time_attr:
                lines.append(time_attr)
            if speed_elem:
                lines.append(speed_elem)
            lines.append('    </trkpt>')
        lines.append('    </trkseg>')
        lines.append('  </trk>')

    lines.append('</gpx>')
    return '\n'.join(lines) + '\n'


def _xml_escape(s: str) -> str:
    return s.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;').replace('"', '&quot;')


def _sanitize_filename(name: str) -> str:
    """Make a trail name safe for use as a filename."""
    return name.replace('/', '-').replace('\\', '-').replace(' ', '_').replace(':', '-')


def main():
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <file.usr> [output_dir]")
        sys.exit(1)

    input_path = sys.argv[1]
    output_dir = sys.argv[2] if len(sys.argv) >= 3 else os.path.dirname(input_path) or '.'

    with open(input_path, 'rb') as f:
        data = f.read()

    trails = parse_usr(data)

    if not trails:
        print("No trails found in file.")
        return

    os.makedirs(output_dir, exist_ok=True)

    for i, trail in enumerate(trails):
        name = trail['name'] or f"trail_{i+1}"
        filename = f"{_sanitize_filename(name)}.gpx"
        output_path = os.path.join(output_dir, filename)

        gpx = trails_to_gpx([trail])
        with open(output_path, 'w') as f:
            f.write(gpx)

        print(f"  -> {output_path} ({len(trail['points'])} points)")


if __name__ == '__main__':
    main()