#!/usr/bin/env python3 """ DCOL GNSS Data Bridge ===================== This module implements a GNSS data bridge for the Trimble DCOL (Data Collection) protocol. It supports GETSVDATA (0x54) retrieval, SETSVDATA (0xA9) uploads, and GPS time synchronization commands so receivers, archives, and analysis tooling can exchange satellite state seamlessly. The goal is to extract GNSS data from one device and feed to a second to test the A9 DCOL support. Key Features: ------------ - Connects to dedicated GET and SEND endpoints simultaneously with optional persistent sockets for mirroring workflows. - Decodes GETSVDATA responses, normalizes payloads, and writes JSON artifacts (and Trimble DAT files when requested). - Sends SETSVDATA payloads for ephemeris, almanac, LAST_KNOWN_POS, and TIME_AND_DATE messages via ``send_time()``. - Bridges multi-subtype pipelines, automatically maps complementary subtypes, and exposes ``run_dcol_operations()`` for scripted integrations. - Provides a command-line interface with extensive flags (routing, subtype selection, PRN ranges, data injection, logging, and debug output). Protocol Overview: ----------------- The DCOL protocol uses two primary commands: 1. 0x54 (GETSVDATA): Request satellite data from receiver (ephemeris, almanac, flags, etc.) 2. 0xA9 (SETSVDATA): Upload satellite data to receiver Packet Structure: ---------------- All packets follow this format: STX (0x02) | STATUS | TYPE | LENGTH | [PAYLOAD] | CHECKSUM | ETX (0x03) Response Structure (0x55 RETSVDATA): STX (0x02) | STATUS | 0x55 | LENGTH | SUBTYPE | SV_PRN | [DATA] | CHECKSUM | ETX (0x03) [PAYLOAD] == SUBTYPE | SV_PRN | [DATA] Supported Satellite Systems: --------------------------- - GPS: Ephemeris (L1/L2/L5, CNAV), Almanac, ION/UTC parameters - Galileo: Ephemeris (E1B/E5A/E5B), Almanac, System Data - GLONASS: Ephemeris, Almanac - QZSS: Ephemeris, Almanac - BeiDou/Compass: Ephemeris, Almanac - SBAS: Ephemeris, Almanac - IRNSS: Ephemeris, Almanac Data Format Notes: ----------------- 1. All multi-byte integers use BIG-ENDIAN byte order ('>H', '>I', '>d') 2. Angular parameters stored as SEMI-CIRCLES (multiply by π for radians) 3. CUC, CUS, CIC, CIS fields are transmitted in semi-circles 4. Each response includes SUBTYPE and SV_PRN_NUMBER in header (not in data payload) 5. Server may respond with NAK (0x15) when no data available Usage Examples: -------------- from dcol_client import ( DCOLClient, SubtypeGetData, SubtypeSendData, run_dcol_operations, ) client = DCOLClient( get_server_ip="192.168.1.100", get_port=6001, send_server_ip="192.168.1.200", send_port=5000, output_dir="dcol_data", debug=True, radians=True, ) ephemeris = client.get_data( subtype=SubtypeGetData.GPS_EPHEMERIS, sv_prn=3, flags=0x00, ) client.send_data( subtype=SubtypeSendData.LAST_KNOWN_POS, data_params={"reserved": 0, "latitude": 37.7749, "longitude": -122.4194, "height": 25.0}, ) client.send_time() run_dcol_operations( get_server_ip="192.168.1.100", get_port=6001, send_server_ip="192.168.1.200", send_port=5000, do_get=True, do_send=True, do_settime=True, subtypes=[SubtypeGetData.GPS_EPHEMERIS], prn="All", output_dir="dcol_data", ) Command-Line Interface: ---------------------- python dcol_client.py --get-server 192.168.1.100 --get-port 6001 --get --subtype 1 --prn 5 --output-dir dcol_data --debug python dcol_client.py --get-server 10.1.150.XXX --get-port 6001 --send-server 192.168.1.200 --send-port 5000 --get --send --subtype 1 --prn All --pos 37.7749,-122.4194 --settime --radians python dcol_client.py --send-server 192.168.1.100 --send-port 5000 --send --send-subtype 0 --data '{"utc_week": 2234, "utc_seconds": 432000}' Additional flags: --get-subtype, --flags, --data-file, --mode, --sat-type, --pos, --datfile Architecture: ------------ - DCOLClient: Main class handling socket communication and disk persistence - run_dcol_operations: Workflow orchestrator for combined GET, SEND, and time operations - SubtypeGetData/SubtypeSendData: Enumerations for command subtypes and payload builders - _decode_*_ephemeris(): Decoders for ephemeris data (GPS, Galileo, GLONASS, etc.) - _decode_*_almanac(): Decoders for almanac data - _parse_response(): Main response parser routing to specific decoders - _format_*(): Convert decoded data to human-readable JSON output Reference Documentation: ----------------------- - GNSS Wiki Copyright: Trimble Inc. 2025 """ import socket import struct import argparse import sys import os import glob import json import math from datetime import datetime, timezone, timedelta from enum import IntEnum from typing import Optional, Dict, Any, Tuple # Only include the data types we want to process. class SubtypeGetData(IntEnum): """Subtypes for 54h GETSVDATA command""" GPS_EPHEMERIS = 1 GPS_ION_UTC = 3 EXTENDED_GPS_ALMANAC = 7 GLONASS_ALMANAC = 8 GLONASS_EPHEMERIS = 9 GALILEO_EPHEMERIS = 11 GALILEO_ALMANAC = 12 QZSS_EPHEMERIS = 14 QZSS_ALMANAC = 16 COMPASS_EPHEMERIS = 21 COMPASS_ALMANAC = 22 BEIDOU_BCNAV_EPHEMERIS = 27 GPS_CNAV_EPHEMERIS = 28 class SubtypeSendData(IntEnum): """Subtypes for A9h send data command""" GPS_ALMANAC = 1 GPS_EPHEMERIS = 2 GPS_IONO_UTC = 3 LAST_KNOWN_POS = 5 GLONASS_ALMANAC = 6 GLONASS_EPHEMERIS = 7 GALILEO_ALMANAC = 11 GALILEO_EPHEMERIS = 12 BEIDOU_ALMANAC = 13 BEIDOU_EPHEMERIS = 14 # Mapping from GET subtype to SEND subtype # Used when both --get and --send are specified with only a GET subtype GET_TO_SEND_SUBTYPE_MAP = { SubtypeGetData.GPS_EPHEMERIS: SubtypeSendData.GPS_EPHEMERIS, # 1 -> 2 SubtypeGetData.GPS_ION_UTC: SubtypeSendData.GPS_IONO_UTC, # 3 -> 3 SubtypeGetData.EXTENDED_GPS_ALMANAC: SubtypeSendData.GPS_ALMANAC, # 7 -> 1 SubtypeGetData.GLONASS_ALMANAC: SubtypeSendData.GLONASS_ALMANAC, # 8 -> 6 SubtypeGetData.GLONASS_EPHEMERIS: SubtypeSendData.GLONASS_EPHEMERIS, # 9 -> 7 SubtypeGetData.GALILEO_EPHEMERIS: SubtypeSendData.GALILEO_EPHEMERIS, # 11 -> 12 SubtypeGetData.GALILEO_ALMANAC: SubtypeSendData.GALILEO_ALMANAC, # 12 -> 11 SubtypeGetData.COMPASS_EPHEMERIS: SubtypeSendData.BEIDOU_EPHEMERIS, # 21 -> 14 SubtypeGetData.COMPASS_ALMANAC: SubtypeSendData.BEIDOU_ALMANAC, # 22 -> 13 } class DCOLClient: """Client for DCOL protocol communication""" # Protocol constants STX = 0x02 ETX = 0x03 ACK = 0x06 NAK = 0x15 CMD_GETSVDATA = 0x54 CMD_RETSVDATA = 0x55 CMD_SENDDATA = 0xA9 CMD_GETTIME = 0x14 CMD_GPSTIME = 0x15 # Display name lookup table - maps native parameter names to human-readable display strings DISPLAY_NAMES = { # Common time parameters 'week': 'Week', 'tow': 'TOW(s)', 'toc': 'TOC(s)', 'toe': 'TOE(s)', 'toa': 'TOA(s)', 'top': 'TOP(s)', # Common identifiers 'sv_number': 'sv_number', 'data_source': 'Data_Source', 'system': 'System', # GPS/QZSS/BeiDou/IRNSS ephemeris parameters 'iodc': 'IODC', 'iode': 'IODE', 'tgd': 'TGD(s)', 'flags': 'Flags', 'reserved': 'Reserved', # Orbital parameters - position 'm0': 'Mean_Anom(sc)', 'ecc': 'Eccentricity', 'sqrta': 'SQRT_A(m^1/2)', 'omega': 'Arg_Perigee(sc)', 'omega0': 'Right_Ascen(sc)', 'i0': 'Inclination(sc)', # Orbital parameters - rates 'idot': 'Inclin_Rate(sc/s)', 'delta_n': 'Mean_Motion_Diff(sc/s)', 'omegadot': 'Right_Ascen_Rate(sc/s)', # Correction terms 'cuc': 'Lat_Cos_Ampl(sc)', 'cus': 'Lat_Sin_Ampl(sc)', 'crs': 'Radius_Cos_Ampl(m)', 'crc': 'Radius_Sin_Ampl(m)', 'cic': 'Inclin_Cos_Ampl(sc)', 'cis': 'Inclin_Sin_Ampl(sc)', # Clock parameters 'af0': 'AF0(s)', 'af1': 'AF1(s/s)', 'af2': 'AF2(s/s/s)', # Almanac parameters 'alm_decode_time': 'ALM_Decode_Time', 'awn': 'Almanac_Week', 'alm_health': 'Health', 'alm_src': 'Source', # Galileo specific 'iodnav': 'IODnav', 'iodalm': 'IODalm', 'bgd1': 'BGD1(s)', 'bgd2': 'BGD2(s)', 'model1': 'Model1', 'model2': 'Model2', 'sisa': 'SISA', 'hsdvs': 'HSDVS', 'health_e5b': 'E5B_Health', 'health_e1': 'E1_Health', 'delta_i': 'Delta_Inclination(sc)', # GLONASS ephemeris 'gps_week_eph_valid': 'GPS_Week_Eph_Valid', 'gps_time_eph_valid': 'GPS_Time_Eph_Valid(s)', 'gps_week_eph_decode': 'GPS_Week_Eph_Decode', 'gps_time_eph_decode': 'GPS_Time_Eph_Decode(s)', 'glonass_day_number': 'GLONASS_Day_Number', 'ref_time_eph': 'Ref_Time_Eph(s)', 'leap_seconds': 'Leap_Seconds', 'frame_start_time': 'Frame_Start_Time', 'age_of_data': 'Age_Of_Data', 'ephemeris_source': 'Ephemeris_Source', 'fdma': 'FDMA_Channel', 'health': 'Health', 'generation': 'Generation', 'udre': 'UDRE', 'x': 'X_Position(km)', 'x_velocity': 'X_Velocity(km/s)', 'x_acceleration': 'X_Acceleration(km/s^2)', 'y': 'Y_Position(km)', 'y_velocity': 'Y_Velocity(km/s)', 'y_acceleration': 'Y_Acceleration(km/s^2)', 'z': 'Z_Position(km)', 'z_velocity': 'Z_Velocity(km/s)', 'z_acceleration': 'Z_Acceleration(km/s^2)', 'a0_utc': 'A0_UTC(s)', 'a0': 'A0(s)', 'a1': 'A1(s/s)', 'tau_gps': 'Tau_GPS(s)', 'delta_tau_n': 'Delta_Tau_n(s)', # GLONASS almanac 'day_number': 'Day_Number', 'fdma_number': 'FDMA_Channel', 'ecc': 'Eccentricity', 'arg_of_perigee': 'Arg_Perigee', 'orbit_period': 'Orbit_Period', 'orbital_period_correction': 'Orbital_Period_Correction', 'long_first_ascending_node': 'Long_First_Ascending_Node', 'time_ascending_node': 'Time_Ascending_Node', 'inclination': 'Inclination', # SBAS ephemeris/almanac 'week_number': 'Week_Number', 'ura': 'URA', 'x_vel': 'X_Velocity(m/s)', 'x_acc': 'X_Acceleration(m/s^2)', 'y_vel': 'Y_Velocity(m/s)', 'y_acc': 'Y_Acceleration(m/s^2)', 'z_vel': 'Z_Velocity(m/s)', 'z_acc': 'Z_Acceleration(m/s^2)', 'validity': 'Validity', # BeiDou B-CNAV 'sat_type': 'Satellite_Type', 'delta_a': 'Delta_A(m)', 'a_dot': 'A_DOT(m/s)', 'delta_n0': 'Delta_n0(sc/s)', 'delta_ndot': 'Delta_n_DOT(sc/s^2)', 'sma': 'Semi_Major_Axis(m)', 'sma_dot': 'SMA_DOT(m/s)', 'delta_n_dot': 'Delta_n_DOT(sc/s^2)', 'tgd_b1cp': 'TGD_B1CP(s)', 'tgd_b2ap': 'TGD_B2AP(s)', 'tgd_b2bi': 'TGD_B2BI(s)', 'isc_b1cd': 'ISC_B1CD(s)', 'isc_b2ad': 'ISC_B2AD(s)', 'sismai': 'SISMAI', 'sisai_oe': 'SISAI_OE', 'sisai_ocb': 'SISAI_OCB', 'sisai_oc1': 'SISAI_OC1', 'sisai_oc2': 'SISAI_OC2', # GPS CNAV 'tow_msg10': 'TOW_MSG10', 'tow_msg11': 'TOW_MSG11', 'uraed_index': 'URAed_Index', 'urai_ned0': 'URAI_NED0', 'urai_ned1': 'URAI_NED1', 'urai_ned2': 'URAI_NED2', 'isc_l1ca': 'ISC_L1CA(s)', 'isc_l2c': 'ISC_L2C(s)', 'isc_l5i5': 'ISC_L5I5(s)', 'isc_l5q5': 'ISC_L5Q5(s)', 'isc_l1cd': 'ISC_L1CD(s)', 'isc_l1cp': 'ISC_L1CP(s)', 'wn_op': 'WN_OP', 'flag1': 'Flag1', 'flag2': 'Flag2', # GPS ION/UTC 'alpha_0': 'Alpha_0', 'alpha_1': 'Alpha_1', 'alpha_2': 'Alpha_2', 'alpha_3': 'Alpha_3', 'beta_0': 'Beta_0', 'beta_1': 'Beta_1', 'beta_2': 'Beta_2', 'beta_3': 'Beta_3', 'asub0': 'A0', 'asub1': 'A1', 'tsub0t': 'T0t', 'deltatls': 'Delta_tLS', 'deltatlsf': 'Delta_tLSF', 'iontime': 'ION_Time', 'wnsubt': 'WNt', 'wnsublsf': 'WN_LSF', 'dn': 'DN', # Galileo system data 'vote_flag': 'Vote_Flag', 'ionization_1st_order': 'Ionization_1st_Order', 'ionization_2nd_order': 'Ionization_2nd_Order', 'ionization_3rd_order': 'Ionization_3rd_Order', 'solar_flux_by_region': 'Solar_Flux_By_Region', 'utc_a0': 'UTC_A0', 'utc_a1': 'UTC_A1', 'delta_t_ls': 'Delta_t_LS', 'utc_tot': 'UTC_TOT', 'utc_wnt': 'UTC_WNT', 'utc_wnlsf': 'UTC_WNLSF', 'utc_dn': 'UTC_DN', 'utc_delta_t_lsf': 'UTC_Delta_t_LSF', 'a0g': 'A0G', 'a1g': 'A1G', 't0g': 'T0G', 'wn0g': 'WN0G', } # Parameters that should be converted from semi-circles to radians when --radians flag is set RADIAN_CONVERTIBLE = { 'm0', 'omega', 'omega0', 'i0', 'idot', 'delta_n', 'omegadot', 'cuc', 'cus', 'cic', 'cis', 'delta_i', 'delta_n0', 'delta_ndot' } def __init__(self, get_server_ip: Optional[str] = None, get_port: Optional[int] = None, send_server_ip: Optional[str] = None, send_port: Optional[int] = None, output_dir: str = "dcol_data", debug: bool = False, radians: bool = False, datfile: Optional[str] = None): """ Initialize DCOL client Args: get_server_ip: Server IP address for GET operations (54h GETSVDATA) get_port: Server port number for GET operations send_server_ip: Server IP address for SEND operations (A9h SENDDATA) send_port: Server port number for SEND operations output_dir: Directory to save received data files debug: Enable debug output radians: Convert semi-circles to radians datfile: Optional path to Trimble DAT file for writing ephemeris records """ self.get_server_ip = get_server_ip self.get_port = get_port self.send_server_ip = send_server_ip self.send_port = send_port self.output_dir = output_dir self.debug = debug self.radians = radians self.data_cache: Dict[str, Any] = {} self.datfile = datfile self.datfile_handle = None self.traffic_log = [] # Track all sent/received traffic for json output # Create output directory if it doesn't exist os.makedirs(self.output_dir, exist_ok=True) # Open DAT file if specified if self.datfile: self.datfile_handle = open(self.datfile, 'wb') if self.debug: print(f"Opened DAT file for writing: {self.datfile}") # ===== Helper methods for decoding common structures ===== def _debug_field(self, name: str, value: Any, data: bytes, offset: int, size: int) -> None: """Print debug output for a field if debug mode is enabled""" if self.debug: hex_bytes = ' '.join(f'{b:02x}' for b in data[offset:offset+size]) print(f" {name}: {hex_bytes} = {value}") def _apply_radians(self, param_name: str, value: float) -> float: """Apply radians conversion at display time if enabled and parameter is convertible""" if self.radians and param_name in self.RADIAN_CONVERTIBLE: return value * math.pi return value def _format_value_for_display(self, param_name: str, value: Any) -> Any: """Format a value for display, applying radian conversion if needed""" if isinstance(value, float) and param_name in self.RADIAN_CONVERTIBLE: return self._apply_radians(param_name, value) return value def _format_for_display(self, decoded_data: Dict[str, Any]) -> Dict[str, Any]: """Convert decoded data to human-readable format for display""" formatted = {} for key, value in decoded_data.items(): display_key = self.DISPLAY_NAMES.get(key, key) display_value = self._format_value_for_display(key, value) formatted[display_key] = display_value return formatted def _print_formatted_data(self, data: Dict[str, Any], indent: int = 0) -> None: """Print data in aligned format with display names""" indent_str = " " * indent for key, value in data.items(): if key == 'decoded' and isinstance(value, dict): print(f"{indent_str}{key}:") # Print decoded data with display names and alignment for field_key, field_value in value.items(): display_key = self.DISPLAY_NAMES.get(field_key, field_key) display_value = self._format_value_for_display(field_key, field_value) # Format with fixed-width column for alignment print(f"{indent_str} {display_key:45s} : {display_value}") elif isinstance(value, dict): print(f"{indent_str}{key}:") self._print_formatted_data(value, indent + 2) elif isinstance(value, (list, tuple)): print(f"{indent_str}{key}: {value}") elif key == 'raw_data' and isinstance(value, str) and len(value) > 100: # Truncate long raw_data hex strings print(f"{indent_str}{key}: {value[:100]}... ({len(value)} chars)") else: print(f"{indent_str}{key}: {value}") def _write_gps_ephemeris_to_dat(self, sv_prn: int, eph: Dict[str, Any]) -> None: """Write GPS ephemeris record to DAT file in Trimble format Args: sv_prn: Satellite PRN number eph: Decoded ephemeris dictionary with native field names Note: Trimble DAT files use little-endian byte order """ if not self.datfile_handle: return # DAT file format for GPS ephemeris (record type 21) # Total record size: 190 bytes record = bytearray() # Preamble (byte): 0x74 record.append(0x74) # Length of record including preamble (byte): 190 record.append(190) # Reserved (byte): 0 record.append(0) # Record type (byte): 21 for GPS ephemeris record.append(21) # SV PRN # (byte) record.append(sv_prn) # Reserved/source (byte): 0 for GPS, or data_source for QZSS source = eph.get('data_source', 0) record.append(source) # WN - Week Number (integer = 2 bytes signed, little-endian) record.extend(struct.pack(' None: """Write Galileo ephemeris record to DAT file in Trimble format Args: sv_prn: Satellite PRN number (SV ID) eph: Decoded ephemeris dictionary with native field names Note: Trimble DAT files use little-endian byte order """ if not self.datfile_handle: return # DAT file format for Galileo ephemeris (record type 28) # Total record size: 188 bytes record = bytearray() # Preamble (byte): 0x74 record.append(0x74) # Length of record including preamble (2 bytes little-endian): 188 record.extend(struct.pack(' Tuple[Dict[str, Any], int]: """ Decode GPS-like ephemeris structure (174 bytes) This decoder is used by GPS, QZSS, Compass/BeiDou, and IRNSS satellites. All follow the ICD-200 GPS ephemeris format with minor variations. Args: data: Raw byte array from RETSVDATA response offset: Starting position in data array has_data_source: If True, includes 1-byte DATA_SOURCE field (QZSS only) Returns: Tuple of (decoded_dict, new_offset) Note: CUC, CUS, CIC, CIS fields are transmitted divided by PI. Multiply by PI (via --radians flag) to get ICD-compliant radians. """ eph = {} start_offset = offset # DATA SOURCE (optional, QZSS only: L1CA=0, L1C=1, L2C=2, L5=3) if has_data_source: eph["data_source"] = struct.unpack_from('B', data, offset)[0] self._debug_field("data_source", eph["data_source"], data, offset, 1) offset += 1 # WEEK NUMBER [weeks] (2 bytes, big-endian) eph["week"] = struct.unpack_from('>H', data, offset)[0] self._debug_field("week", eph["week"], data, offset, 2) offset += 2 # IODC - Issue of Data, Clock (2 bytes, big-endian) eph["iodc"] = struct.unpack_from('>H', data, offset)[0] self._debug_field("iodc", eph["iodc"], data, offset, 2) offset += 2 # RESERVED (1 byte, always 0) eph["reserved"] = struct.unpack_from('B', data, offset)[0] self._debug_field("reserved", eph["reserved"], data, offset, 1) offset += 1 # IODE - Issue of Data, Ephemeris (1 byte) eph["iode"] = struct.unpack_from('B', data, offset)[0] self._debug_field("iode", eph["iode"], data, offset, 1) offset += 1 # TOW - Time of Week [seconds] (4 bytes signed, big-endian) eph["tow"] = struct.unpack_from('>l', data, offset)[0] self._debug_field("tow", eph["tow"], data, offset, 4) offset += 4 # TOC - Clock Data Reference Time [seconds] (4 bytes signed, big-endian) eph["toc"] = struct.unpack_from('>l', data, offset)[0] self._debug_field("toc", eph["toc"], data, offset, 4) offset += 4 # TOE - Ephemeris Reference Time [seconds] (4 bytes signed, big-endian) eph["toe"] = struct.unpack_from('>l', data, offset)[0] self._debug_field("toe", eph["toe"], data, offset, 4) offset += 4 # TGD - Group Delay Differential [seconds] (8 bytes double, big-endian) eph["tgd"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("tgd", eph["tgd"], data, offset, 8) offset += 8 # AF2 - Clock Drift Rate [sec/sec^2] (8 bytes double, big-endian) eph["af2"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("af2", eph["af2"], data, offset, 8) offset += 8 # AF1 - Clock Drift [sec/sec] (8 bytes double, big-endian) eph["af1"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("af1", eph["af1"], data, offset, 8) offset += 8 # AF0 - Clock Bias [seconds] (8 bytes double, big-endian) eph["af0"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("af0", eph["af0"], data, offset, 8) offset += 8 # CRS - Sine Harmonic Correction to Orbit Radius [meters] (8 bytes double) eph["crs"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("crs", eph["crs"], data, offset, 8) offset += 8 # DELTA_N - Mean Motion Difference [semi-circles/sec] (8 bytes double) eph["delta_n"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("delta_n", eph["delta_n"], data, offset, 8) offset += 8 # M0 - Mean Anomaly at Reference Time [semi-circles] (8 bytes double) eph["m0"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("m0", eph["m0"], data, offset, 8) offset += 8 # CUC - Cosine Harmonic Correction to Arg of Latitude [radians/PI] (8 bytes double) # Note: Transmitted as CUC/PI, multiply by PI for ICD units eph["cuc"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("cuc", eph["cuc"], data, offset, 8) offset += 8 # ECCENTRICITY - Orbital Eccentricity [dimensionless] (8 bytes double) eph["ecc"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("ecc", eph["ecc"], data, offset, 8) offset += 8 # CUS - Sine Harmonic Correction to Arg of Latitude [radians/PI] (8 bytes double) # Note: Transmitted as CUS/PI, multiply by PI for ICD units eph["cus"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("cus", eph["cus"], data, offset, 8) offset += 8 # SQRT_A - Square Root of Semi-Major Axis [meters^1/2] (8 bytes double) eph["sqrta"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("sqrta", eph["sqrta"], data, offset, 8) offset += 8 # CIC - Cosine Harmonic Correction to Inclination [radians/PI] (8 bytes double) # Note: Transmitted as CIC/PI, multiply by PI for ICD units eph["cic"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("cic", eph["cic"], data, offset, 8) offset += 8 # OMEGA0 - Longitude of Ascending Node [semi-circles] (8 bytes double) eph["omega0"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("omega0", eph["omega0"], data, offset, 8) offset += 8 # CIS - Sine Harmonic Correction to Inclination [radians/PI] (8 bytes double) # Note: Transmitted as CIS/PI, multiply by PI for ICD units eph["cis"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("cis", eph["cis"], data, offset, 8) offset += 8 # I0 - Inclination Angle at Reference Time [semi-circles] (8 bytes double) eph["i0"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("i0", eph["i0"], data, offset, 8) offset += 8 # CRC - Cosine Harmonic Correction to Orbit Radius [meters] (8 bytes double) eph["crc"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("crc", eph["crc"], data, offset, 8) offset += 8 # OMEGA - Argument of Perigee [semi-circles] (8 bytes double) eph["omega"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("omega", eph["omega"], data, offset, 8) offset += 8 # OMEGA_DOT - Rate of Right Ascension [semi-circles/sec] (8 bytes double) eph["omegadot"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("omegadot", eph["omegadot"], data, offset, 8) offset += 8 # IDOT - Rate of Inclination Angle [semi-circles/sec] (8 bytes double) eph["idot"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("idot", eph["idot"], data, offset, 8) offset += 8 # FLAGS - Status and configuration flags (4 bytes unsigned, big-endian) eph["flags"] = struct.unpack_from('>L', data, offset)[0] self._debug_field("flags", eph["flags"], data, offset, 4) offset += 4 return eph, offset def _decode_gps_like_almanac(self, data: bytes, offset: int, has_health_16bit: bool = False, has_clock_params: bool = True, has_alm_src: bool = True) -> Tuple[Dict[str, Any], int]: """ Decode GPS-like almanac structure (67-85 bytes) Used by GPS, QZSS, Compass/BeiDou, IRNSS, and Galileo almanacs. All follow a similar structure based on the GPS almanac format. Args: data: Raw byte array from RETSVDATA response offset: Starting position in data array has_health_16bit: If True, health is 2 bytes (BeiDou); if False, 1 byte (GPS/others) has_clock_params: If True, includes AF0/AF1 clock parameters (Extended Almanac subtype 7) Returns: Tuple of (decoded_dict, new_offset) Note: Angular parameters stored as semi-circles, convert with --radians flag """ alm = {} start_offset = offset # ALM_DECODE_TIME - Time when almanac was decoded [GPS seconds] (4 bytes unsigned, big-endian) alm["alm_decode_time"] = struct.unpack_from('>L', data, start_offset)[0] self._debug_field("alm_decode_time", alm["alm_decode_time"], data, start_offset, 4) # AWN - Almanac Week Number [weeks] (2 bytes unsigned, big-endian) alm["awn"] = struct.unpack_from('>H', data, start_offset + 4)[0] self._debug_field("awn", alm["awn"], data, start_offset + 4, 2) # TOA - Time of Almanac [seconds] (4 bytes unsigned, big-endian) alm["toa"] = struct.unpack_from('>L', data, start_offset + 6)[0] self._debug_field("toa", alm["toa"], data, start_offset + 6, 4) # SQRT_A - Square Root of Semi-Major Axis [meters^1/2] (8 bytes double, big-endian) alm["sqrta"] = struct.unpack_from('>d', data, start_offset + 10)[0] self._debug_field("sqrta", alm["sqrta"], data, start_offset + 10, 8) # ECCENTRICITY - Orbital Eccentricity [dimensionless] (8 bytes double, big-endian) alm["ecc"] = struct.unpack_from('>d', data, start_offset + 18)[0] self._debug_field("ecc", alm["ecc"], data, start_offset + 18, 8) # I0 - Inclination Angle [semi-circles] (8 bytes double, big-endian) alm["i0"] = struct.unpack_from('>d', data, start_offset + 26)[0] self._debug_field("i0", alm["i0"], data, start_offset + 26, 8) # OMEGA_DOT - Rate of Right Ascension [semi-circles/sec] (8 bytes double, big-endian) alm["omegadot"] = struct.unpack_from('>d', data, start_offset + 34)[0] self._debug_field("omegadot", alm["omegadot"], data, start_offset + 34, 8) # OMEGA0 - Longitude of Ascending Node [semi-circles] (8 bytes double, big-endian) alm["omega0"] = struct.unpack_from('>d', data, start_offset + 42)[0] self._debug_field("omega0", alm["omega0"], data, start_offset + 42, 8) # OMEGA - Argument of Perigee [semi-circles] (8 bytes double, big-endian) alm["omega"] = struct.unpack_from('>d', data, start_offset + 50)[0] self._debug_field("omega", alm["omega"], data, start_offset + 50, 8) # M0 - Mean Anomaly [semi-circles] (8 bytes double, big-endian) alm["m0"] = struct.unpack_from('>d', data, start_offset + 58)[0] self._debug_field("m0", alm["m0"], data, start_offset + 58, 8) # ALM_HEALTH - Satellite health status (1 or 2 bytes depending on system) # BeiDou uses 2 bytes (9-bit health), others use 1 byte (8-bit health) if has_health_16bit: alm["alm_health"] = struct.unpack_from('>H', data, start_offset + 66)[0] self._debug_field("alm_health", alm["alm_health"], data, start_offset + 66, 2) offset = start_offset + 68 else: alm["alm_health"] = struct.unpack_from('B', data, start_offset + 66)[0] self._debug_field("alm_health", alm["alm_health"], data, start_offset + 66, 1) offset = start_offset + 67 # Clock parameters only in extended almanac formats (subtype 7 GPS Extended, Galileo, QZSS, etc.) if has_clock_params: # AF0 - Clock Bias [seconds] (8 bytes double, big-endian) alm["af0"] = struct.unpack_from('>d', data, offset)[0] self._debug_field("af0", alm["af0"], data, offset, 8) alm["af1"] = struct.unpack_from('>d', data, offset + 8)[0] self._debug_field("af1", alm["af1"], data, offset + 8, 8) offset += 16 # Alm-src only for Galileo and QZSS, not for GPS Extended if has_alm_src: alm["alm_src"] = struct.unpack_from('B', data, offset)[0] self._debug_field("alm_src", alm["alm_src"], data, offset, 1) offset += 1 return alm, offset def _decode_galileo_ephemeris(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode Galileo ephemeris data (183 bytes total, similar to GPS format) Note: 55.txt doesn't document SV_NUMBER, but actual response includes it (like GPS) """ gal_eph = {} # SV_NUMBER (not in 55.txt, but present in actual response) gal_eph["sv_number"] = data[offset] self._debug_field("sv_number", gal_eph["sv_number"], data, offset, 1) offset += 1 # DATA SOURCE (E1B:0 / E5B:1 / E5A:2) gal_eph["data_source"] = data[offset] self._debug_field("data_source", gal_eph["data_source"], data, offset, 1) offset += 1 # WEEK NUMBER (GST, with 1024 added for normal Galileo satellites) gal_eph["week"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("week", gal_eph["week"], data, offset, 2) offset += 2 # TOW (GST) gal_eph["tow"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("tow", gal_eph["tow"], data, offset, 4) offset += 4 # IODnav (Ephemeris and Clock Correction Issue of Data) gal_eph["iodnav"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("iodnav", gal_eph["iodnav"], data, offset, 2) offset += 2 # TOE [seconds] gal_eph["toe"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("toe", gal_eph["toe"], data, offset, 4) offset += 4 # CRS [meters] gal_eph["crs"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("crs", gal_eph["crs"], data, offset, 8) offset += 8 # DELTAN [semi-circles / sec] gal_eph["delta_n"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("delta_n", gal_eph["delta_n"], data, offset, 8) offset += 8 # MSUB0 [semi-circles] gal_eph["m0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("m0", gal_eph["m0"], data, offset, 8) offset += 8 # CUC [semi-circles] (ICD in radians) gal_eph["cuc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("cuc", gal_eph["cuc"], data, offset, 8) offset += 8 # ECCENT [dimensionless] gal_eph["ecc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("ecc", gal_eph["ecc"], data, offset, 8) offset += 8 # CUS [semi-circles] (ICD in radians) gal_eph["cus"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("cus", gal_eph["cus"], data, offset, 8) offset += 8 # SQRTA [sqrt(meters)] gal_eph["sqrta"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("sqrta", gal_eph["sqrta"], data, offset, 8) offset += 8 # CIC [semi-circles] (ICD in radians) gal_eph["cic"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("cic", gal_eph["cic"], data, offset, 8) offset += 8 # OMEGSUB0 [semi-circles] gal_eph["omega0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omega0", gal_eph["omega0"], data, offset, 8) offset += 8 # CIS [semi-circles] (ICD in radians) gal_eph["cis"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("cis", gal_eph["cis"], data, offset, 8) offset += 8 # ISUB0 [semi-circles] gal_eph["i0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("i0", gal_eph["i0"], data, offset, 8) offset += 8 # CRC [meters] gal_eph["crc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("crc", gal_eph["crc"], data, offset, 8) offset += 8 # OMEGA [semi-circles] gal_eph["omega"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omega", gal_eph["omega"], data, offset, 8) offset += 8 # OMEGADOT [semi-circles / sec] gal_eph["omegadot"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omegadot", gal_eph["omegadot"], data, offset, 8) offset += 8 # IDOT [semi-circles / sec] gal_eph["idot"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("idot", gal_eph["idot"], data, offset, 8) offset += 8 # SISA (1 byte) gal_eph["sisa"] = data[offset] self._debug_field("sisa", gal_eph["sisa"], data, offset, 1) offset += 1 # HSDVS (Signal Health Flag) - 2 bytes # Bit 0 = E5A DVS, bit 2-1 = E5A HS, bit 3 = E5B DVS, bit 5-4 = E5B HS, bit 6 = E1B DVS, bit 8-7 = E1B HS gal_eph["hsdvs"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("hsdvs", gal_eph["hsdvs"], data, offset, 2) offset += 2 # TOC (E1, E5B) for source E1B/E5B (E1, E5A) for source E5A gal_eph["toc"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("toc", gal_eph["toc"], data, offset, 4) offset += 4 # AF0 [s] gal_eph["af0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af0", gal_eph["af0"], data, offset, 8) offset += 8 # AF1 [s/s] gal_eph["af1"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af1", gal_eph["af1"], data, offset, 8) offset += 8 # AF2 [s/s^2] gal_eph["af2"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af2", gal_eph["af2"], data, offset, 8) offset += 8 # BGD1 [s] gal_eph["bgd1"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("bgd1", gal_eph["bgd1"], data, offset, 8) offset += 8 # MODEL1 (Clock model for TOC/AF0~2/BGD1) gal_eph["model1"] = data[offset] self._debug_field("model1", gal_eph["model1"], data, offset, 1) offset += 1 # BGD2 [s] gal_eph["bgd2"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("bgd2", gal_eph["bgd2"], data, offset, 8) offset += 8 # MODEL2 (Clock model for BGD2) gal_eph["model2"] = data[offset] self._debug_field("model2", gal_eph["model2"], data, offset, 1) offset += 1 return gal_eph, offset def _decode_galileo_almanac(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode Galileo almanac data (87 bytes total: 1 subtype + 86 payload) Structure: sv_number, alm_decode_time, awn, toa, sqrta, ecc, i0, omegadot, omega0, omega, m0, alm_health, af0, af1, alm_src, iodalm """ gal_alm = {} # SV_NUMBER (1 byte) - not in 55.txt but present in actual response gal_alm["sv_number"] = data[offset] self._debug_field("sv_number", gal_alm["sv_number"], data, offset, 1) offset += 1 # ALM_DECODE_TIME (4 bytes unsigned) gal_alm["alm_decode_time"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("alm_decode_time", gal_alm["alm_decode_time"], data, offset, 4) offset += 4 # AWN - Almanac Week Number (2 bytes) gal_alm["awn"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("awn", gal_alm["awn"], data, offset, 2) offset += 2 # TOA - Time of Almanac (4 bytes unsigned) gal_alm["toa"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("toa", gal_alm["toa"], data, offset, 4) offset += 4 # SQRTA (8 bytes double) gal_alm["sqrta"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("sqrta", gal_alm["sqrta"], data, offset, 8) offset += 8 # ECCENTRICITY (8 bytes double) gal_alm["ecc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("ecc", gal_alm["ecc"], data, offset, 8) offset += 8 # I0 (ISUBO in 55.txt) - Inclination (8 bytes double) gal_alm["i0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("i0", gal_alm["i0"], data, offset, 8) offset += 8 # OMEGADOT (8 bytes double) gal_alm["omegadot"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omegadot", gal_alm["omegadot"], data, offset, 8) offset += 8 # OMEGA0 (OMEGSUBO in 55.txt) - Longitude of Ascending Node (8 bytes double) gal_alm["omega0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omega0", gal_alm["omega0"], data, offset, 8) offset += 8 # OMEGA - Argument of Perigee (8 bytes double) gal_alm["omega"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("omega", gal_alm["omega"], data, offset, 8) offset += 8 # M0 (MSUBO in 55.txt) - Mean Anomaly (8 bytes double) gal_alm["m0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("m0", gal_alm["m0"], data, offset, 8) offset += 8 # ALM_HEALTH (1 byte) gal_alm["alm_health"] = data[offset] self._debug_field("alm_health", gal_alm["alm_health"], data, offset, 1) offset += 1 # AF0 (ASUBF0 in 55.txt) - Clock Bias (8 bytes double) gal_alm["af0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af0", gal_alm["af0"], data, offset, 8) offset += 8 # AF1 (ASUBF1 in 55.txt) - Clock Drift (8 bytes double) gal_alm["af1"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af1", gal_alm["af1"], data, offset, 8) offset += 8 # ALM_SRC - Almanac Source (1 byte) gal_alm["alm_src"] = data[offset] self._debug_field("alm_src", gal_alm["alm_src"], data, offset, 1) offset += 1 # IODalm - Issue of Data Almanac (1 byte) gal_alm["iodalm"] = data[offset] self._debug_field("iodalm", gal_alm["iodalm"], data, offset, 1) offset += 1 return gal_alm, offset def _decode_glonass_ephemeris(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode GLONASS ephemeris data (139 bytes)""" glo_eph = {} glo_eph["gps_week_eph_valid"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("gps_week_eph_valid", glo_eph["gps_week_eph_valid"], data, offset, 2) offset += 2 glo_eph["gps_time_eph_valid"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("gps_time_eph_valid", glo_eph["gps_time_eph_valid"], data, offset, 4) offset += 4 glo_eph["gps_week_eph_decode"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("gps_week_eph_decode", glo_eph["gps_week_eph_decode"], data, offset, 2) offset += 2 glo_eph["gps_time_eph_decode"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("gps_time_eph_decode", glo_eph["gps_time_eph_decode"], data, offset, 4) offset += 4 glo_eph["glonass_day_number"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("glonass_day_number", glo_eph["glonass_day_number"], data, offset, 2) offset += 2 glo_eph["ref_time_eph"] = data[offset] self._debug_field("ref_time_eph", glo_eph["ref_time_eph"], data, offset, 1) offset += 1 glo_eph["leap_seconds"] = data[offset] self._debug_field("leap_seconds", glo_eph["leap_seconds"], data, offset, 1) offset += 1 glo_eph["flags"] = data[offset] self._debug_field("flags", glo_eph["flags"], data, offset, 1) offset += 1 glo_eph["frame_start_time"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("frame_start_time", glo_eph["frame_start_time"], data, offset, 4) offset += 4 glo_eph["age_of_data"] = data[offset] self._debug_field("age_of_data", glo_eph["age_of_data"], data, offset, 1) offset += 1 glo_eph["ephemeris_source"] = data[offset] self._debug_field("ephemeris_source", glo_eph["ephemeris_source"], data, offset, 1) offset += 1 glo_eph["fdma"] = struct.unpack('b', data[offset:offset+1])[0] # signed self._debug_field("fdma", glo_eph["fdma"], data, offset, 1) offset += 1 glo_eph["health"] = data[offset] self._debug_field("health", glo_eph["health"], data, offset, 1) offset += 1 glo_eph["generation"] = data[offset] self._debug_field("generation", glo_eph["generation"], data, offset, 1) offset += 1 glo_eph["udre"] = data[offset] self._debug_field("udre", glo_eph["udre"], data, offset, 1) offset += 1 glo_eph["x"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x", glo_eph["x"], data, offset, 8) offset += 8 glo_eph["x_velocity"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x_velocity", glo_eph["x_velocity"], data, offset, 8) offset += 8 glo_eph["x_acceleration"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x_acceleration", glo_eph["x_acceleration"], data, offset, 8) offset += 8 glo_eph["y"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y", glo_eph["y"], data, offset, 8) offset += 8 glo_eph["y_velocity"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y_velocity", glo_eph["y_velocity"], data, offset, 8) offset += 8 glo_eph["y_acceleration"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y_acceleration", glo_eph["y_acceleration"], data, offset, 8) offset += 8 glo_eph["z"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z", glo_eph["z"], data, offset, 8) offset += 8 glo_eph["z_velocity"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z_velocity", glo_eph["z_velocity"], data, offset, 8) offset += 8 glo_eph["z_acceleration"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z_acceleration", glo_eph["z_acceleration"], data, offset, 8) offset += 8 glo_eph["a0_utc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("a0_utc", glo_eph["a0_utc"], data, offset, 8) offset += 8 glo_eph["a0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("a0", glo_eph["a0"], data, offset, 8) offset += 8 glo_eph["a1"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("a1", glo_eph["a1"], data, offset, 8) offset += 8 glo_eph["tau_gps"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("tau_gps", glo_eph["tau_gps"], data, offset, 8) offset += 8 glo_eph["delta_tau_n"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("delta_tau_n", glo_eph["delta_tau_n"], data, offset, 8) offset += 8 return glo_eph, offset def _decode_glonass_almanac(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode GLONASS almanac data (68 bytes)""" glo_alm = {} glo_alm["day_number"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("day_number", glo_alm["day_number"], data, offset, 2) offset += 2 glo_alm["fdma_number"] = struct.unpack('b', data[offset:offset+1])[0] # signed self._debug_field("fdma_number", glo_alm["fdma_number"], data, offset, 1) offset += 1 glo_alm["ecc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("ecc", glo_alm["ecc"], data, offset, 8) offset += 8 glo_alm["arg_of_perigee"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("arg_of_perigee", glo_alm["arg_of_perigee"], data, offset, 8) offset += 8 glo_alm["orbit_period"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("orbit_period", glo_alm["orbit_period"], data, offset, 8) offset += 8 glo_alm["orbital_period_correction"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("orbital_period_correction", glo_alm["orbital_period_correction"], data, offset, 8) offset += 8 glo_alm["long_first_ascending_node"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("long_first_ascending_node", glo_alm["long_first_ascending_node"], data, offset, 8) offset += 8 glo_alm["time_ascending_node"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("time_ascending_node", glo_alm["time_ascending_node"], data, offset, 8) offset += 8 glo_alm["inclination"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("inclination", glo_alm["inclination"], data, offset, 8) offset += 8 glo_alm["a0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("a0", glo_alm["a0"], data, offset, 8) offset += 8 glo_alm["health"] = data[offset] self._debug_field("health", glo_alm["health"], data, offset, 1) offset += 1 return glo_alm, offset def _decode_sbas_ephemeris(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode SBAS ephemeris data (96 bytes)""" sbas_eph = {} sbas_eph["week_number"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("week_number", sbas_eph["week_number"], data, offset, 2) offset += 2 sbas_eph["toe"] = struct.unpack('>i', data[offset:offset+4])[0] # signed self._debug_field("toe", sbas_eph["toe"], data, offset, 4) offset += 4 sbas_eph["iode"] = data[offset] self._debug_field("iode", sbas_eph["iode"], data, offset, 1) offset += 1 sbas_eph["ura"] = data[offset] self._debug_field("ura", sbas_eph["ura"], data, offset, 1) offset += 1 sbas_eph["x"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x", sbas_eph["x"], data, offset, 8) offset += 8 sbas_eph["x_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x_vel", sbas_eph["x_vel"], data, offset, 8) offset += 8 sbas_eph["x_acc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x_acc", sbas_eph["x_acc"], data, offset, 8) offset += 8 sbas_eph["y"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y", sbas_eph["y"], data, offset, 8) offset += 8 sbas_eph["y_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y_vel", sbas_eph["y_vel"], data, offset, 8) offset += 8 sbas_eph["y_acc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y_acc", sbas_eph["y_acc"], data, offset, 8) offset += 8 sbas_eph["z"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z", sbas_eph["z"], data, offset, 8) offset += 8 sbas_eph["z_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z_vel", sbas_eph["z_vel"], data, offset, 8) offset += 8 sbas_eph["z_acc"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z_acc", sbas_eph["z_acc"], data, offset, 8) offset += 8 sbas_eph["af0"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af0", sbas_eph["af0"], data, offset, 8) offset += 8 sbas_eph["af1"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("af1", sbas_eph["af1"], data, offset, 8) offset += 8 return sbas_eph, offset def _decode_sbas_almanac(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode SBAS almanac data (56 bytes)""" sbas_alm = {} sbas_alm["week"] = struct.unpack('>H', data[offset:offset+2])[0] self._debug_field("week", sbas_alm["week"], data, offset, 2) offset += 2 sbas_alm["toe"] = struct.unpack('>I', data[offset:offset+4])[0] self._debug_field("toe", sbas_alm["toe"], data, offset, 4) offset += 4 sbas_alm["x"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x", sbas_alm["x"], data, offset, 8) offset += 8 sbas_alm["y"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y", sbas_alm["y"], data, offset, 8) offset += 8 sbas_alm["z"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z", sbas_alm["z"], data, offset, 8) offset += 8 sbas_alm["x_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("x_vel", sbas_alm["x_vel"], data, offset, 8) offset += 8 sbas_alm["y_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("y_vel", sbas_alm["y_vel"], data, offset, 8) offset += 8 sbas_alm["z_vel"] = struct.unpack('>d', data[offset:offset+8])[0] self._debug_field("z_vel", sbas_alm["z_vel"], data, offset, 8) offset += 8 sbas_alm["validity"] = data[offset] self._debug_field("validity", sbas_alm["validity"], data, offset, 1) offset += 1 sbas_alm["health"] = data[offset] self._debug_field("health", sbas_alm["health"], data, offset, 1) offset += 1 return sbas_alm, offset def _decode_gps_ion_utc(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode GPS ION/UTC data (123 bytes)""" ion = {} ion["reserved"] = struct.unpack_from('B', data, offset)[0] offset += 1 ion["alpha_0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["alpha_1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["alpha_2"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["alpha_3"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["beta_0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["beta_1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["beta_2"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["beta_3"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["asub0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["asub1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["tsub0t"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["deltatls"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["deltatlsf"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["iontime"] = struct.unpack_from('>d', data, offset)[0] offset += 8 ion["wnsubt"] = struct.unpack_from('B', data, offset)[0] offset += 1 ion["wnsublsf"] = struct.unpack_from('B', data, offset)[0] offset += 1 ion["dn"] = struct.unpack_from('B', data, offset)[0] offset += 1 return ion, offset def _decode_galileo_sys_data(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode Galileo System Data (70 bytes)""" gal_sys = {} gal_sys["data_source"] = struct.unpack_from('B', data, offset)[0] offset += 1 gal_sys["vote_flag"] = struct.unpack_from('B', data, offset)[0] offset += 1 gal_sys["week"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["tow"] = struct.unpack_from('>L', data, offset)[0] offset += 4 gal_sys["ionization_1st_order"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["ionization_2nd_order"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["ionization_3rd_order"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["solar_flux_by_region"] = struct.unpack_from('B', data, offset)[0] offset += 1 gal_sys["utc_a0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["utc_a1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["delta_t_ls"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["utc_tot"] = struct.unpack_from('>L', data, offset)[0] offset += 4 gal_sys["utc_wnt"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["utc_wnlsf"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["utc_dn"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["utc_delta_t_lsf"] = struct.unpack_from('>H', data, offset)[0] offset += 2 gal_sys["a0g"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["a1g"] = struct.unpack_from('>d', data, offset)[0] offset += 8 gal_sys["t0g"] = struct.unpack_from('>L', data, offset)[0] offset += 4 gal_sys["wn0g"] = struct.unpack_from('>H', data, offset)[0] offset += 2 return gal_sys, offset def _decode_beidou_bcnav_ephemeris(self, data: bytes, offset: int) -> Tuple[Dict[str, Any], int]: """Decode BeiDou B-CNAV ephemeris (233 bytes)""" eph = {} eph["data_source"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["week"] = struct.unpack_from('>H', data, offset)[0] offset += 2 eph["tow"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["toe"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["sat_type"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["reserved"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["iode"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["delta_a"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["a_dot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["delta_n0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["delta_ndot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["m0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["ecc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omega"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omega0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["i0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omegadot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["idot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cis"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cic"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["crs"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["crc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cuc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cus"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["toc"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["iodc"] = struct.unpack_from('>H', data, offset)[0] offset += 2 eph["af0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["af1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["af2"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["tgd_b1cp"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["tgd_b2ap"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["tgd_b2bi"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_b1cd"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_b2ad"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["top"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["sismai"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["sisai_oe"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["sisai_ocb"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["sisai_oc1"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["sisai_oc2"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["flags"] = struct.unpack_from('>L', data, offset)[0] offset += 4 return eph, offset def _decode_gps_cnav_ephemeris(self, data: bytes, offset: int, include_extended: bool = False) -> Tuple[Dict[str, Any], int]: """Decode GPS CNAV ephemeris (158 bytes base, 247 bytes extended)""" eph = {} eph["data_source"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["week"] = struct.unpack_from('>H', data, offset)[0] offset += 2 eph["top"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["toe"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["tow_msg10"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["tow_msg11"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["sma"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["sma_dot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["delta_n"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["delta_n_dot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["m0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["ecc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omega"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omega0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["omegadot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["i0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["idot"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cis"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cic"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["crs"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["crc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cus"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["cuc"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["uraed_index"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["flag1"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["flag2"] = struct.unpack_from('B', data, offset)[0] offset += 1 # Extended fields (GPS_CNAV_EXTENDED only) if include_extended: eph["toc"] = struct.unpack_from('>l', data, offset)[0] offset += 4 eph["af0"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["af1"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["af2"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["urai_ned0"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["urai_ned1"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["urai_ned2"] = struct.unpack_from('B', data, offset)[0] offset += 1 eph["tgd"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_l1ca"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_l2c"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_l5i5"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_l5q5"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["wn_op"] = struct.unpack_from('>H', data, offset)[0] offset += 2 eph["isc_l1cd"] = struct.unpack_from('>d', data, offset)[0] offset += 8 eph["isc_l1cp"] = struct.unpack_from('>d', data, offset)[0] offset += 8 return eph, offset def _calculate_checksum(self, data: bytes) -> int: """Calculate checksum for DCOL protocol (Sum of all bytes MOD 256)""" checksum = 0 for byte in data: checksum += byte return checksum % 256 def _build_packet(self, cmd_type: int, payload: bytes) -> bytes: """ Build a DCOL protocol packet Args: cmd_type: Command type byte payload: Command payload Returns: Complete packet with STX, STATUS, TYPE, LENGTH, payload, CHECKSUM, ETX """ status = 0x00 length = len(payload) # Build packet: STX + STATUS + TYPE + LENGTH + payload packet_data = bytes([status, cmd_type, length]) + payload # Calculate checksum on STATUS through end of payload checksum = self._calculate_checksum(packet_data) # Complete packet packet = bytes([self.STX]) + packet_data + bytes([checksum, self.ETX]) # Output packet as hex bytes hex_output = ' '.join(f'{b:02x}' for b in packet) print(f"Packet ({len(packet)} bytes): {hex_output}") return packet def _log_traffic(self, direction: str, endpoint: str, data: bytes, description: str = "") -> None: """Log traffic for later export to JSON""" entry = { "timestamp": datetime.now().isoformat(), "direction": direction, # "sent" or "received" "endpoint": endpoint, # "get" or "send" "description": description, "hex": data.hex(), "size_bytes": len(data) } self.traffic_log.append(entry) def _send_packet(self, packet: bytes, sock: socket.socket, endpoint: str = "unknown", description: str = "") -> None: """Send a packet to the server""" self._log_traffic("sent", endpoint, packet, description) sock.sendall(packet) def _receive_response(self, sock: socket.socket, endpoint: str = "unknown", timeout: float = 5.0) -> Optional[bytes]: """ Receive a response from the server Returns: Response payload (without STX, STATUS, TYPE, LENGTH, CHECKSUM, ETX) or None """ sock.settimeout(timeout) try: # Read first byte (STX or possibly NAK) first_byte = sock.recv(1) if not first_byte: print("Error: No response from server") return None # Check if it's a NAK (no data available) if first_byte[0] == self.NAK: self._log_traffic("received", endpoint, first_byte, "NAK response") print("Server responded with NAK (no data available)") return None # Check if it's STX if first_byte[0] != self.STX: print(f"Error: Invalid STX received: {first_byte[0]:02x}") return None # Read STATUS, TYPE, LENGTH header = sock.recv(3) if len(header) != 3: print("Error: Incomplete header") return None status, cmd_type, length = struct.unpack('BBB', header) # Read payload payload = b'' while len(payload) < length: chunk = sock.recv(length - len(payload)) if not chunk: print("Error: Connection closed while reading payload") return None payload += chunk # Read checksum and ETX footer = sock.recv(2) if len(footer) != 2: print("Error: Incomplete footer") return None checksum_received, etx = struct.unpack('BB', footer) if etx != self.ETX: print(f"Error: Invalid ETX: {etx:02x}") return None # Verify checksum checksum_calc = self._calculate_checksum(header + payload) if checksum_calc != checksum_received: print(f"Error: Checksum mismatch. Calc: {checksum_calc:02x}, Recv: {checksum_received:02x}") return None # Log the complete received packet full_packet = first_byte + header + payload + footer self._log_traffic("received", endpoint, full_packet, f"Response: type={cmd_type:02x}, length={length}") # Note: We don't check for NAK here because 0x15 is also a valid command type (GPSTIME) # NAK checking is only relevant for A9h SENDDATA responses (single byte response) return payload except socket.timeout: print("Error: Timeout waiting for response") return None except Exception as e: print(f"Error receiving response: {e}") return None def get_data(self, subtype: int, sv_prn: int = 0, flags: int = 0, sat_type: int = 0, mode: int = 0, sock: Optional[socket.socket] = None) -> Optional[Dict[str, Any]]: """ Get satellite data from server (54h GETSVDATA command) Args: subtype: Subtype indicating what data is requested (see SubtypeGetData) sv_prn: Satellite PRN number (ignored for some subtypes) flags: Flags field (bitmapped) sat_type: Satellite type (for subtype 20 only) mode: Mode (for subtype 20 only) sock: Optional existing socket connection (for persistent mode) Returns: Dictionary containing parsed response data, or None on error """ if not self.get_server_ip or not self.get_port: print("Error: GET server IP and port not configured") return None print(f"Requesting data from {self.get_server_ip}:{self.get_port}: subtype={subtype}, sv_prn={sv_prn}, flags={flags:#04x}") # Build payload based on subtype payload = struct.pack('BBB', subtype, sv_prn, flags) # Build and send packet packet = self._build_packet(self.CMD_GETSVDATA, payload) # Determine if we need to manage the socket or use provided one close_socket = (sock is None) try: if sock is None: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((self.get_server_ip, self.get_port)) self._send_packet(packet, sock, "get", f"GETSVDATA: subtype={subtype}, sv_prn={sv_prn}") # Receive response (should be RETSVDATA 55h) response = self._receive_response(sock, "get") if response is None: return None # Parse response based on subtype parsed_data = self._parse_response(subtype, response) # Add SV_PRN to decoded data if not already present (for almanacs that don't include it) if parsed_data and 'decoded' in parsed_data: if 'sv_number' not in parsed_data['decoded'] and 'sv_number' not in parsed_data['decoded']: parsed_data['decoded']['sv_number'] = sv_prn # Save to file if parsed_data: self._save_data_to_file(subtype, sv_prn, parsed_data) # Store in cache cache_key = f"subtype_{subtype}_prn_{sv_prn}" self.data_cache[cache_key] = parsed_data return parsed_data except ConnectionRefusedError: print(f"Error: Connection refused to {self.get_server_ip}:{self.get_port}") return None except Exception as e: print(f"Error in get_data: {e}") return None finally: if close_socket and sock: sock.close() def get_time(self, uptime: bool = False) -> Optional[Dict[str, Any]]: """ Get GPS time from receiver using GETTIME (14h) command Args: uptime: If True, request system uptime instead of GPS time Returns: Dictionary with time information or None on error """ try: print('Requesting time from server...') # Build GETTIME packet (14h) if uptime: payload = bytes([0x01]) # Request uptime else: payload = bytes() # Request GPS time (no parameter or value != 1) packet = self._build_packet(self.CMD_GETTIME, payload) if self.debug: packet_hex = ' '.join(f'{b:02x}' for b in packet) print(f"Packet ({len(packet)} bytes): {packet_hex}") # Send request and receive response with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.connect((self.get_server_ip, self.get_port)) self._send_packet(packet, sock, "get", "GETTIME request") # Receive response (should be GPSTIME 15h) response = self._receive_response(sock, "get") if response is None: return None # Parse GPSTIME response result = self._parse_gpstime_response(response, uptime) return result except ConnectionRefusedError: print(f"Error: Connection refused to {self.get_server_ip}:{self.get_port}") return None except Exception as e: print(f"Error in get_time: {e}") return None def _parse_gpstime_response(self, data: bytes, uptime_requested: bool) -> Dict[str, Any]: """Parse GPSTIME (15h) response""" result = {} if len(data) < 1: return result if uptime_requested: # Response is 4 bytes: UPTIME in seconds if len(data) >= 4: uptime_seconds = struct.unpack('>L', data[0:4])[0] result["uptime_seconds"] = uptime_seconds result["uptime_days"] = uptime_seconds / 86400.0 if self.debug: print(f"System uptime: {uptime_seconds} seconds ({result['uptime_days']:.2f} days)") else: # Response format: 7 char TIME, 4 char WEEK, 2 char GPS/UTC OFFSET, 4 char TIME OFFSET, 3 char TIME MODE # Total: 20 bytes if len(data) >= 20: # TIME: 7 characters (ASCII) - seconds since beginning of GPS week time_str = data[0:7].decode('ascii', errors='ignore').strip() result["time_seconds"] = int(time_str) if time_str.isdigit() else 0 # WEEK NUMBER: 4 characters (ASCII) - GPS week number week_str = data[7:11].decode('ascii', errors='ignore').strip() result["week_number"] = int(week_str) if week_str.isdigit() else 0 # GPS/UTC OFFSET: 2 characters (ASCII) - leap seconds offset_str = data[11:13].decode('ascii', errors='ignore').strip() result["gps_utc_offset"] = int(offset_str) if offset_str.lstrip('-').isdigit() else 0 # TIME OFFSET: 4 characters (ASCII) - local time zone in signed minutes time_offset_str = data[13:17].decode('ascii', errors='ignore').strip() result["time_offset_minutes"] = int(time_offset_str) if time_offset_str.lstrip('-').isdigit() else 0 # TIME MODE: 3 characters (ASCII) - 'UTC', 'L24', or 'L12' time_mode_str = data[17:20].decode('ascii', errors='ignore').strip() result["time_mode"] = time_mode_str # Calculate UTC seconds (GPS time - leap seconds) result["utc_seconds"] = result["time_seconds"] - result["gps_utc_offset"] if self.debug: print(f"GPS Time: Week {result['week_number']}, Seconds {result['time_seconds']}") print(f"GPS/UTC Offset: {result['gps_utc_offset']} seconds") print(f"UTC Time: Week {result['week_number']}, Seconds {result['utc_seconds']}") print(f"Time Offset: {result['time_offset_minutes']} minutes") print(f"Time Mode: {result['time_mode']}") return result def _parse_response(self, subtype: int, data: bytes) -> Dict[str, Any]: result = {"subtype": subtype, "raw_data": data.hex(), "length": len(data)} try: if subtype == SubtypeGetData.GPS_EPHEMERIS: result["data_type"] = "GPS Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 175: # Full ephemeris (1 byte SV + 174 bytes data after subtype) sv_number = struct.unpack_from('B', data, offset)[0] self._debug_field("sv_number", sv_number, data, offset, 1) offset += 1 # Use common GPS-like ephemeris decoder eph, offset = self._decode_gps_like_ephemeris(data, offset, has_data_source=False) eph["sv_number"] = sv_number # Add SV number to ephemeris dict # Format output using common formatter eph["system"] = "GPS" result["decoded"].update(eph) elif subtype == SubtypeGetData.EXTENDED_GPS_ALMANAC: result["data_type"] = "GPS Almanac Extended" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 84: # GPS Almanac Extended is 84 bytes including SV_NUMBER (1 + 83) sv_number = struct.unpack_from('B', data, offset)[0] self._debug_field("sv_number", sv_number, data, offset, 1) offset += 1 # GPS Extended Almanac has clock params but no alm_src field alm, offset = self._decode_gps_like_almanac(data, offset, has_health_16bit=False, has_clock_params=True, has_alm_src=False) # Only output formatted names alm["sv_number"] = sv_number result["decoded"].update(alm) elif subtype == SubtypeGetData.GPS_ION_UTC: result["data_type"] = "GPS ION/UTC" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 # ION/UTC data is 122 bytes after subtype byte (1 reserved + 14*8 doubles + 3 bytes) if len(data) >= offset + 122: ion, offset = self._decode_gps_ion_utc(data, offset) result["decoded"].update(ion) else: if self.debug: print(f"GPS ION/UTC: Insufficient data. Need {offset + 122} bytes, have {len(data)}") elif subtype == SubtypeGetData.GLONASS_ALMANAC: result["data_type"] = "GLONASS Almanac" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 68: # GLONASS Almanac is 68 bytes (BIG ENDIAN) alm, offset = self._decode_glonass_almanac(data, offset) alm["system"] = "GLONASS" result["decoded"].update(alm) elif subtype == SubtypeGetData.GLONASS_EPHEMERIS: result["data_type"] = "GLONASS Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 139: # GLONASS Ephemeris is 139 bytes (BIG ENDIAN) eph, offset = self._decode_glonass_ephemeris(data, offset) eph["system"] = "GLONASS" result["decoded"].update(eph) elif subtype == SubtypeGetData.GALILEO_EPHEMERIS: result["data_type"] = "Galileo Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 183: # Galileo Ephemeris is 183 bytes (including SV_NUMBER not in 55.txt) eph, offset = self._decode_galileo_ephemeris(data, offset) eph["system"] = "Galileo" result["decoded"].update(eph) elif subtype == SubtypeGetData.GALILEO_ALMANAC: result["data_type"] = "Galileo Almanac" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 85: # Galileo Almanac is 85 bytes (BIG ENDIAN) alm, offset = self._decode_galileo_almanac(data, offset) alm["system"] = "Galileo" result["decoded"].update(alm) elif subtype == SubtypeGetData.QZSS_EPHEMERIS: result["data_type"] = "QZSS Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 175: # QZSS Ephemeris is 175 bytes (BIG ENDIAN) eph, offset = self._decode_gps_like_ephemeris(data, offset, has_data_source=True) # Only output formatted names eph["system"] = "QZSS" result["decoded"].update(eph) elif subtype == SubtypeGetData.QZSS_ALMANAC: result["data_type"] = "QZSS Almanac" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 84: # QZSS Almanac is 84 bytes (BIG ENDIAN) alm, offset = self._decode_gps_like_almanac(data, offset, has_health_16bit=False) # Only output formatted names alm["system"] = "QZSS" result["decoded"].update(alm) elif subtype == SubtypeGetData.COMPASS_EPHEMERIS: result["data_type"] = "Compass/BeiDou Ephemeris" if len(data) >= 2: # Need at least subtype + sv_number offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 # BeiDou includes SV_NUMBER in response (undocumented in 55.txt but present in actual data) result["decoded"]["sv_number"] = struct.unpack_from('B', data, offset)[0] self._debug_field("sv_number", result["decoded"]["sv_number"], data, offset, 1) offset += 1 if len(data) >= offset + 174: # Compass Ephemeris is 174 bytes (BIG ENDIAN) eph, offset = self._decode_gps_like_ephemeris(data, offset, has_data_source=False) # Only output formatted names eph["system"] = "Compass/BeiDou" result["decoded"].update(eph) elif subtype == SubtypeGetData.COMPASS_ALMANAC: result["data_type"] = "Compass/BeiDou Almanac" if len(data) >= 2: # Need at least subtype + sv_number offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 # BeiDou includes SV_NUMBER in response (undocumented in 55.txt but present in actual data) result["decoded"]["sv_number"] = struct.unpack_from('B', data, offset)[0] self._debug_field("sv_number", result["decoded"]["sv_number"], data, offset, 1) offset += 1 if len(data) >= offset + 85: # Compass Almanac is 85 bytes (BIG ENDIAN, 16-bit health) alm, offset = self._decode_gps_like_almanac(data, offset, has_health_16bit=True) # Only output formatted names alm["system"] = "Compass/BeiDou" result["decoded"].update(alm) elif subtype == SubtypeGetData.BEIDOU_BCNAV_EPHEMERIS: result["data_type"] = "BeiDou B-CNAV Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 233: # BeiDou B-CNAV is 233 bytes (BIG ENDIAN) eph, offset = self._decode_beidou_bcnav_ephemeris(data, offset) eph["system"] = "BeiDou B-CNAV" result["decoded"].update(eph) elif subtype == SubtypeGetData.GPS_CNAV_EPHEMERIS: result["data_type"] = "GPS CNAV Ephemeris" if len(data) >= 1: offset = 0 result["decoded"] = { "subtype_response": struct.unpack_from('B', data, offset)[0] } offset += 1 if len(data) >= offset + 158: # GPS CNAV is 158 bytes (BIG ENDIAN) eph, offset = self._decode_gps_cnav_ephemeris(data, offset, include_extended=False) eph["system"] = "GPS CNAV" result["decoded"].update(eph) else: result["data_type"] = f"Subtype {subtype}" # For unknown subtypes, just note the subtype byte if present if len(data) >= 1: result["decoded"] = { "subtype_response": struct.unpack_from('B', data, 0)[0] } except Exception as e: result["parse_error"] = str(e) return result def _save_data_to_file(self, subtype: int, sv_prn: int, data: Dict[str, Any]) -> None: """Save received data to a JSON file with native field names""" filename = os.path.join(self.output_dir, f"subtype_{subtype}_prn_{sv_prn}.json") # Save data with native field names (no formatting) with open(filename, 'w') as f: json.dump(data, f, indent=2) print(f"Data saved to: {filename}") def send_data(self, subtype: int, data_params: Dict[str, Any], sock: Optional[socket.socket] = None) -> bool: """ Send satellite data to server (A9h command) Args: subtype: Subtype indicating what data is being sent (see SubtypeSendData) data_params: Dictionary containing the data parameters for the subtype sock: Optional existing socket connection (for persistent mode) Returns: True if ACK received, False otherwise """ if not self.send_server_ip or not self.send_port: print("Error: SEND server IP and port not configured") return False print(f"Sending data to {self.send_server_ip}:{self.send_port}: subtype={subtype}") # Build payload based on subtype try: payload = self._build_send_payload(subtype, data_params) except Exception as e: print(f"Error building payload: {e}") return False if payload is None: print(f"Error: Unsupported subtype {subtype}") return False # Build and send packet packet = self._build_packet(self.CMD_SENDDATA, payload) # Determine if we need to manage the socket or use provided one close_socket = (sock is None) try: if sock is None: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((self.send_server_ip, self.send_port)) self._send_packet(packet, sock, "send", f"SETSVDATA: subtype={subtype}") # Wait for ACK or NAK sock.settimeout(5.0) response = sock.recv(1) if not response: print("Error: No response from server") return False # Log the ACK/NAK response self._log_traffic("received", "send", response, f"ACK/NAK response: 0x{response[0]:02x}") if response[0] == self.ACK: print("Success: Received ACK") return True elif response[0] == self.NAK: print("Error: Received NAK") return False else: print(f"Error: Unexpected response: {response[0]:02x}") return False except ConnectionRefusedError: print(f"Error: Connection refused to {self.send_server_ip}:{self.send_port}") return False except Exception as e: print(f"Error in send_data: {e}") return False finally: if close_socket and sock: sock.close() def send_time(self, sock: Optional[socket.socket] = None) -> bool: """ Send TIME_AND_DATE (A9:0) message to send server using PC system time Args: sock: Optional existing socket connection (for persistent mode) Returns: True if ACK received, False otherwise """ if not self.send_server_ip or not self.send_port: print("Error: SEND server IP and port not configured") return False if self.debug: print(f"Sending time to {self.send_server_ip}:{self.send_port}") # --- Setup --- gps_epoch = datetime(1980, 1, 6, tzinfo=timezone.utc) current_utc = datetime.now(timezone.utc) # --- Calculation --- # Get the difference (returns a timedelta object) diff = current_utc - gps_epoch # Convert the difference to total raw seconds total_elapsed_seconds = diff.total_seconds() # Calculate GPS Week and Time of Week (TOW) SECONDS_IN_WEEK = 604800 week = int(total_elapsed_seconds // SECONDS_IN_WEEK) tow = int(total_elapsed_seconds % SECONDS_IN_WEEK) # Remainder seconds if self.debug: print(f" GPS Week: {week}, Time of Week: {tow} seconds") print(f" UTC Time: {current_utc.isoformat()}") # Build payload: Subtype 0 for TIME_AND_DATE payload = struct.pack('B', 0) # Subtype 0 # Note per the documentation this is in UTC, there's no # adjustment for leap seconds! payload += struct.pack('>HL', week, tow) # Big-endian: week (2 bytes), tow (4 bytes) # Build packet packet = self._build_packet(self.CMD_SENDDATA, payload) # Determine if we need to manage the socket or use provided one close_socket = (sock is None) try: if sock is None: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((self.send_server_ip, self.send_port)) self._send_packet(packet, sock, "send", "SETSVDATA: TIME_AND_DATE (subtype=0)") # Wait for response (should be ACK 0x06 or possibly GPSTIME 0x15) sock.settimeout(5.0) response = sock.recv(1024) # Allow for larger response (GPSTIME packet) if not response: print("Error: No response from server") return False # Log the response if len(response) == 1: # Single byte response (ACK or NAK) self._log_traffic("received", "send", response, f"ACK/NAK response: 0x{response[0]:02x}") if response[0] == self.ACK: if self.debug: print(" Success: Received ACK") print("Time sent successfully.") return True elif response[0] == self.NAK: print("Error: Received NAK") return False else: if self.debug: print(f" Unexpected single-byte response: 0x{response[0]:02x}") else: # Multi-byte response (possibly GPSTIME 0x15 packet) self._log_traffic("received", "send", response, f"GPSTIME or multi-byte response ({len(response)} bytes)") if self.debug: hex_output = ' '.join(f'{b:02x}' for b in response) print(f" Received multi-byte response ({len(response)} bytes): {hex_output}") # Check if it's a valid packet with ACK or contains STX if response[0] == self.STX: # It's a packet response (e.g., GPSTIME) if self.debug: print(" Received GPSTIME packet response") print("Time sent successfully (GPSTIME response).") return True elif self.ACK in response: # ACK found somewhere in response if self.debug: print(" Found ACK in response") print("Time sent successfully.") return True else: print(f"Error: Unexpected response format") return False return False except socket.timeout: print("Error: Timeout waiting for response") return False except Exception as e: print(f"Error sending time: {e}") return False finally: if close_socket and sock: try: sock.shutdown(socket.SHUT_RDWR) except: pass sock.close() def _build_send_payload(self, subtype: int, params: Dict[str, Any]) -> Optional[bytes]: """Build payload for send_data based on subtype""" payload = struct.pack('B', subtype) if subtype == SubtypeSendData.GPS_ALMANAC: # Subtype 1: GPS Almanac (84 bytes) (BIG ENDIAN) payload += struct.pack('B', params['sv_number']) payload += struct.pack('>L', params['alm_decode_time']) payload += struct.pack('>H', params['awn']) payload += struct.pack('>L', params['toa']) payload += struct.pack('>d', params['sqrta']) payload += struct.pack('>d', params['ecc']) payload += struct.pack('>d', params['i0']) payload += struct.pack('>d', params['omegadot']) payload += struct.pack('>d', params['omega0']) payload += struct.pack('>d', params['omega']) payload += struct.pack('>d', params['m0']) payload += struct.pack('B', params['alm_health']) payload += struct.pack('>d', params['af0']) payload += struct.pack('>d', params['af1']) elif subtype == SubtypeSendData.GPS_EPHEMERIS: # Subtype 2: GPS Ephemeris (174 bytes after SV number) (BIG ENDIAN) # Structure matches get_GPS_Beidou_QZSS_hh_ephemeris() in iorx_svData.c payload += struct.pack('B', params['sv_number']) # 1 byte payload += struct.pack('>H', params['week']) # 2 bytes - Week payload += struct.pack('>H', params['iodc']) # 2 bytes - IODC # IODE packed into lower byte of U16 (upper byte reserved/ignored) iode_packed = params['iode'] & 0xFF payload += struct.pack('>H', iode_packed) # 2 bytes - IODE in lower byte payload += struct.pack('>l', int(params['tow'])) # 4 bytes - t_ephem payload += struct.pack('>l', int(params['toc'])) # 4 bytes - t_oc payload += struct.pack('>l', int(params['toe'])) # 4 bytes - t_oe payload += struct.pack('>d', params['tgd']) # 8 bytes - T_GD payload += struct.pack('>d', params['af2']) # 8 bytes - a_f2 payload += struct.pack('>d', params['af1']) # 8 bytes - a_f1 payload += struct.pack('>d', params['af0']) # 8 bytes - a_f0 payload += struct.pack('>d', params['crs']) # 8 bytes - C_rs # Note: Server multiplies these by PI, so we send as semi-circles (value/PI) payload += struct.pack('>d', params['delta_n']) # 8 bytes - delta_n (will be *PI on server) payload += struct.pack('>d', params['m0']) # 8 bytes - M_0 (will be *PI on server) payload += struct.pack('>d', params['cuc']) # 8 bytes - C_uc (will be *PI on server) payload += struct.pack('>d', params['ecc']) # 8 bytes - e payload += struct.pack('>d', params['cus']) # 8 bytes - C_us (will be *PI on server) payload += struct.pack('>d', params['sqrta']) # 8 bytes - sqrt_A payload += struct.pack('>d', params['cic']) # 8 bytes - C_ic (will be *PI on server) payload += struct.pack('>d', params['omega0']) # 8 bytes - OMEGA_0 (will be *PI on server) payload += struct.pack('>d', params['cis']) # 8 bytes - C_is (will be *PI on server) payload += struct.pack('>d', params['i0']) # 8 bytes - i_0 (will be *PI on server) payload += struct.pack('>d', params['crc']) # 8 bytes - C_rc payload += struct.pack('>d', params['omega']) # 8 bytes - omega (will be *PI on server) payload += struct.pack('>d', params['omegadot']) # 8 bytes - OMEGADOT (will be *PI on server) payload += struct.pack('>d', params['idot']) # 8 bytes - IDOT (will be *PI on server) payload += struct.pack('>L', params['flags']) # 4 bytes - Flags # Total: 1 (subtype) + 1 (SV) + 2+2+2+4+4+4 (18 headers) + 8*19 (152 doubles) + 4 (flags) = 176 bytes elif subtype == SubtypeSendData.GPS_IONO_UTC: # Subtype 3: GPS Iono/UTC (BIG ENDIAN) payload += struct.pack('B', params.get('reserved', 0)) payload += struct.pack('>d', params['alpha_0']) payload += struct.pack('>d', params['alpha_1']) payload += struct.pack('>d', params['alpha_2']) payload += struct.pack('>d', params['alpha_3']) payload += struct.pack('>d', params['beta_0']) payload += struct.pack('>d', params['beta_1']) payload += struct.pack('>d', params['beta_2']) payload += struct.pack('>d', params['beta_3']) payload += struct.pack('>d', params['asub0']) payload += struct.pack('>d', params['asub1']) payload += struct.pack('>d', params['tsub0t']) payload += struct.pack('>d', params['deltatls']) payload += struct.pack('>d', params['deltatlsf']) payload += struct.pack('>d', params['iontime']) payload += struct.pack('B', params['wnsubt']) payload += struct.pack('B', params['wnsublsf']) payload += struct.pack('B', params['dn']) payload += bytes(6) # 6 reserved bytes elif subtype == SubtypeSendData.LAST_KNOWN_POS: # Subtype 5: Last Known Position (BIG ENDIAN) payload += struct.pack('B', params.get('reserved', 0)) payload += struct.pack('>d', params['latitude']) payload += struct.pack('>d', params['longitude']) payload += struct.pack('>d', params['height']) elif subtype == SubtypeSendData.GALILEO_EPHEMERIS: # Subtype 12: Galileo Ephemeris (183 bytes) (BIG ENDIAN) payload += struct.pack('B', params['sv_number']) # 1 byte payload += struct.pack('B', params['data_source']) # 1 byte - E1B:0 / E5B:1 / E5A:2 payload += struct.pack('>H', params['week']) # 2 bytes - GST week (with 1024 added) payload += struct.pack('>L', params['tow']) # 4 bytes - GST TOW payload += struct.pack('>H', params['iodnav']) # 2 bytes - IODnav payload += struct.pack('>L', params['toe']) # 4 bytes - TOE payload += struct.pack('>d', params['crs']) # 8 bytes - CRS payload += struct.pack('>d', params['delta_n']) # 8 bytes - DELTAN (will be *PI on server) payload += struct.pack('>d', params['m0']) # 8 bytes - MSUB0 (will be *PI on server) payload += struct.pack('>d', params['cuc']) # 8 bytes - CUC/PI (will be *PI on server) payload += struct.pack('>d', params['ecc']) # 8 bytes - ECCENT payload += struct.pack('>d', params['cus']) # 8 bytes - CUS/PI (will be *PI on server) payload += struct.pack('>d', params['sqrta']) # 8 bytes - SQRTA payload += struct.pack('>d', params['cic']) # 8 bytes - CIC/PI (will be *PI on server) payload += struct.pack('>d', params['omega0']) # 8 bytes - OMEGSUB0 (will be *PI on server) payload += struct.pack('>d', params['cis']) # 8 bytes - CIS/PI (will be *PI on server) payload += struct.pack('>d', params['i0']) # 8 bytes - ISUB0 (will be *PI on server) payload += struct.pack('>d', params['crc']) # 8 bytes - CRC payload += struct.pack('>d', params['omega']) # 8 bytes - OMEGA (will be *PI on server) payload += struct.pack('>d', params['omegadot']) # 8 bytes - OMEGADOT (will be *PI on server) payload += struct.pack('>d', params['idot']) # 8 bytes - IDOT (will be *PI on server) payload += struct.pack('B', params['sisa']) # 1 byte - SISA payload += struct.pack('>H', params['hsdvs']) # 2 bytes - HSDVS payload += struct.pack('>L', params['toc']) # 4 bytes - TOC payload += struct.pack('>d', params['af0']) # 8 bytes - AF0 payload += struct.pack('>d', params['af1']) # 8 bytes - AF1 payload += struct.pack('>d', params['af2']) # 8 bytes - AF2 payload += struct.pack('>d', params['bgd1']) # 8 bytes - BGD1 payload += struct.pack('B', params['model1']) # 1 byte - MODEL1 payload += struct.pack('>d', params['bgd2']) # 8 bytes - BGD2 payload += struct.pack('B', params['model2']) # 1 byte - MODEL2 # Total: 1 (subtype) + 1 (SV) + 1 (data_source) + 2+4+2+4 (12 headers) + 8*18 (144 doubles) + 1+2+4 (7 more) + 8+8+8+8+1+8+1 (42 clock) = 184 bytes elif subtype == SubtypeSendData.GALILEO_ALMANAC: # Subtype 11: Galileo Almanac (86 bytes) (BIG ENDIAN) payload += struct.pack('B', params['sv_number']) # 1 byte payload += struct.pack('>L', params['alm_decode_time']) # 4 bytes payload += struct.pack('>H', params['awn']) # 2 bytes payload += struct.pack('>L', params['toa']) # 4 bytes payload += struct.pack('>d', params['sqrta']) # 8 bytes payload += struct.pack('>d', params['ecc']) # 8 bytes payload += struct.pack('>d', params['i0']) # 8 bytes - Delta_I (will be *PI on server) payload += struct.pack('>d', params['omegadot']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['omega0']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['omega']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['m0']) # 8 bytes (will be *PI on server) payload += struct.pack('B', params['alm_health']) # 1 byte payload += struct.pack('>d', params['af0']) # 8 bytes - ASUBF0 payload += struct.pack('>d', params['af1']) # 8 bytes - ASUBF1 payload += struct.pack('B', params.get('alm_src', 0)) # 1 byte - Alm-src (for Galileo) payload += struct.pack('B', params['iodalm']) # 1 byte - IODalm (only for Galileo) # Total: 1 (subtype) + 1 (SV) + 4+2+4 (10) + 8*9 (72) + 1+8+8+1+1 (19) = 87 bytes elif subtype == SubtypeSendData.BEIDOU_EPHEMERIS: # Subtype 14: BeiDou Ephemeris (175 bytes) (BIG ENDIAN) # Structure is same as GPS ephemeris (subtype 2) but without DATA_SOURCE field payload += struct.pack('B', params['sv_number']) # 1 byte payload += struct.pack('>H', params['week']) # 2 bytes - Week (GPS time) payload += struct.pack('>H', params['iodc']) # 2 bytes - IODC # IODE packed into lower byte of U16 (upper byte reserved/ignored) iode_packed = params['iode'] & 0xFF payload += struct.pack('>H', iode_packed) # 2 bytes - IODE in lower byte payload += struct.pack('>l', int(params['tow'])) # 4 bytes - TOW (GPS time) payload += struct.pack('>l', int(params['toc'])) # 4 bytes - TOC (GPS time) payload += struct.pack('>l', int(params['toe'])) # 4 bytes - TOE (GPS time) payload += struct.pack('>d', params['tgd']) # 8 bytes - TGD payload += struct.pack('>d', params['af2']) # 8 bytes - AF2 payload += struct.pack('>d', params['af1']) # 8 bytes - AF1 payload += struct.pack('>d', params['af0']) # 8 bytes - AF0 payload += struct.pack('>d', params['crs']) # 8 bytes - CRS # Note: Server multiplies these by PI, so we send as semi-circles (value/PI) payload += struct.pack('>d', params['delta_n']) # 8 bytes - DELTA_N (will be *PI on server) payload += struct.pack('>d', params['m0']) # 8 bytes - M_0 (will be *PI on server) payload += struct.pack('>d', params['cuc']) # 8 bytes - CUC (will be *PI on server) payload += struct.pack('>d', params['ecc']) # 8 bytes - ECCENTRICITY payload += struct.pack('>d', params['cus']) # 8 bytes - CUS (will be *PI on server) payload += struct.pack('>d', params['sqrta']) # 8 bytes - SQRT_A payload += struct.pack('>d', params['cic']) # 8 bytes - CIC (will be *PI on server) payload += struct.pack('>d', params['omega0']) # 8 bytes - OMEGA_0 (will be *PI on server) payload += struct.pack('>d', params['cis']) # 8 bytes - CIS (will be *PI on server) payload += struct.pack('>d', params['i0']) # 8 bytes - I_0 (will be *PI on server) payload += struct.pack('>d', params['crc']) # 8 bytes - CRC payload += struct.pack('>d', params['omega']) # 8 bytes - OMEGA (will be *PI on server) payload += struct.pack('>d', params['omegadot']) # 8 bytes - OMEGADOT (will be *PI on server) payload += struct.pack('>d', params['idot']) # 8 bytes - IDOT (will be *PI on server) payload += struct.pack('>L', params['flags']) # 4 bytes - FLAGS (includes TGD2 for BeiDou) # Total: 1 (subtype) + 1 (SV) + 2+2+2+4+4+4 (18 headers) + 8*19 (152 doubles) + 4 (flags) = 176 bytes elif subtype == SubtypeSendData.BEIDOU_ALMANAC: # Subtype 13: BeiDou Almanac (86 bytes) (BIG ENDIAN) # Structure similar to GPS almanac but with 16-bit health and alm_src payload += struct.pack('B', params['sv_number']) # 1 byte payload += struct.pack('>L', params['alm_decode_time']) # 4 bytes payload += struct.pack('>H', params['awn']) # 2 bytes payload += struct.pack('>L', params['toa']) # 4 bytes payload += struct.pack('>d', params['sqrta']) # 8 bytes payload += struct.pack('>d', params['ecc']) # 8 bytes payload += struct.pack('>d', params['i0']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['omegadot']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['omega0']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['omega']) # 8 bytes (will be *PI on server) payload += struct.pack('>d', params['m0']) # 8 bytes (will be *PI on server) payload += struct.pack('>H', params['alm_health']) # 2 bytes - 16-bit health for BeiDou payload += struct.pack('>d', params['af0']) # 8 bytes - ASUBF0 payload += struct.pack('>d', params['af1']) # 8 bytes - ASUBF1 payload += struct.pack('B', params.get('alm_src', 0)) # 1 byte - Alm-src (BeiDou has this) # Note: No IODalm for BeiDou (only Galileo has it) # Total: 1 (subtype) + 1 (SV) + 4+2+4 (10) + 8*8 (64) + 2 (health) + 8+8 (16) + 1 (alm_src) = 86 bytes else: print(f"Warning: Subtype {subtype} payload building not fully implemented") # Return basic payload with just subtype return payload return payload def validate_subtype(subtype: int, is_get: bool) -> None: """ Validate that a subtype is supported. Args: subtype: The subtype value to validate is_get: True if this is a GET operation, False if SEND Raises: ValueError: If the subtype is not supported """ if is_get: valid_subtypes = [e.value for e in SubtypeGetData] operation = "GET" else: valid_subtypes = [e.value for e in SubtypeSendData] operation = "SEND" if subtype not in valid_subtypes: raise ValueError( f"Unsupported {operation} subtype: {subtype}. " f"Valid {operation} subtypes are: {', '.join(str(s) for s in sorted(valid_subtypes))}" ) def get_prn_range_for_subtype(get_subtype: Optional[int], send_subtype: Optional[int]) -> tuple: """ Determine the PRN range for a given subtype based on constellation. Args: get_subtype: GET operation subtype (55h RETSVDATA) send_subtype: SEND operation subtype (A9h SENDDATA) Returns: Tuple of (min_prn, max_prn) for the constellation """ # Determine which subtype to use (prefer GET, fallback to SEND) subtype = get_subtype if get_subtype is not None else send_subtype if subtype is None: return (1, 32) # Default to GPS range # GPS subtypes (GET: 1, 3, 7, 28; SEND: 2, 3) if subtype in [1, 2, 3, 7, 28]: # GPS ephemeris/iono/almanac/CNAV return (1, 32) # GLONASS subtypes (GET: 8, 9; SEND: 6, 7) elif subtype in [6, 7, 8, 9]: # GLONASS almanac/ephemeris return (1, 24) # Galileo subtypes (GET: 11, 12; SEND: 11, 12) elif subtype in [11, 12]: # Galileo ephemeris/almanac return (1, 36) # BeiDou/Compass subtypes (GET: 21, 22, 27; SEND: 13, 14) elif subtype in [13, 14, 21, 22, 27]: # BeiDou almanac/ephemeris return (1, 63) # QZSS subtypes (GET: 14, 16; no SEND support) elif subtype in [14, 16]: # QZSS ephemeris/almanac return (193, 202) # Default to GPS range return (1, 32) def run_dcol_operations( get_server_ip: Optional[str] = None, get_port: Optional[int] = None, send_server_ip: Optional[str] = None, send_port: Optional[int] = None, do_get: bool = False, do_settime: bool = False, do_send: bool = False, subtypes: Optional[list] = None, prn: Any = 0, flags: int = 0, sat_type: int = 0, mode: int = 0, send_subtype: Optional[int] = None, data_params: Optional[Dict[str, Any]] = None, data_file: Optional[str] = None, output_dir: str = "dcol_data", debug: bool = False, radians: bool = False, datfile: Optional[str] = None, position_params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Run DCOL operations programmatically Args: get_server_ip: Server IP address for GET operations get_port: Server port number for GET operations send_server_ip: Server IP address for SEND operations send_port: Server port number for SEND operations do_get: Perform GET operation (54h GETSVDATA) do_settime: Sets the time on the "send" receiver do_send: Perform SEND operation (A9h SETSVDATA) subtypes: List of subtype integers (e.g., [1, 11, 12]) prn: PRN number (integer) or "All" for all satellites flags: Flags field for GET operations sat_type: Satellite type (for subtype 20) mode: Mode (for subtype 20) send_subtype: Explicit SEND subtype (overrides auto-mapping) data_params: Dictionary with data parameters for SEND data_file: Path to JSON file with data parameters for SEND output_dir: Output directory for saved data debug: Enable debug output radians: Convert semi-circles to radians datfile: Path to Trimble DAT file for ephemeris records position_params: Dictionary with position data (latitude, longitude, height) for LAST_KNOWN_POS Returns: Dictionary with results: { 'success': bool, 'get_results': list, # List of GET results for each subtype/PRN 'time_result': dict, # Reserved for future GETTIME support 'send_results': list, # List of SEND success flags 'success_count': int, 'fail_count': int, 'traffic_log': list, # Traffic log entries 'traffic_file': str # Path to traffic.json if written } """ # Validation if not do_get and not do_send and not do_settime: raise ValueError("At least one of do_get, do_settime, or do_send must be True") if do_settime and (not send_server_ip or not send_port): raise ValueError("do_settime requires send_server_ip and send_port") if do_get and (not get_server_ip or not get_port): raise ValueError("do_get requires get_server_ip and get_port") if do_send and (not send_server_ip or not send_port): raise ValueError("do_send requires send_server_ip and send_port") # Handle subtypes if subtypes is None: subtypes = [] elif not isinstance(subtypes, list): subtypes = [subtypes] if do_get and not subtypes: raise ValueError("do_get requires subtypes list") # Validate subtypes for st in subtypes: if do_get and st == 0: if do_send: raise ValueError("GET subtype 0 (SV Flags) is incompatible with SEND operations") get_subtype = subtypes[0] if subtypes else None # Parse PRN parameter if str(prn).lower() == 'all': prn_list_mode = True min_prn, max_prn = get_prn_range_for_subtype(get_subtype if do_get else None, send_subtype if do_send else None) prn_list = list(range(min_prn, max_prn + 1)) else: prn_list_mode = False try: prn_value = int(prn) prn_list = [prn_value] except (ValueError, TypeError): raise ValueError(f"prn must be an integer or 'All', got: {prn}") # Create client client = DCOLClient( get_server_ip=get_server_ip, get_port=get_port, send_server_ip=send_server_ip, send_port=send_port, output_dir=output_dir, debug=debug, radians=radians, datfile=datfile ) get_results = [] time_result = None send_results = [] # Open persistent connections if in multi mode get_sock = None send_sock = None multi_mode = prn_list_mode or len(subtypes) > 1 if multi_mode: try: if do_get: get_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) get_sock.connect((get_server_ip, get_port)) if do_send or do_settime: send_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) send_sock.connect((send_server_ip, send_port)) except Exception as e: if get_sock: get_sock.close() if send_sock: send_sock.close() raise ConnectionError(f"Error opening persistent connections: {e}") # Process each subtype success_count = 0 fail_count = 0 sent_time = False sent_position = False try: # Send position first if provided and --send is specified if do_send and position_params: if debug: print(f"Sending LAST_KNOWN_POS: lat={position_params['latitude']}, lon={position_params['longitude']}") pos_success = client.send_data(SubtypeSendData.LAST_KNOWN_POS, position_params, sock=send_sock) send_results.append({ 'subtype': SubtypeSendData.LAST_KNOWN_POS, 'prn': 0, # No PRN for position 'success': pos_success }) if pos_success: success_count += 1 print("Position sent successfully.") else: fail_count += 1 print("Failed to send position.") sent_position = True # If the user requests do_settime, we sent it early in the sequence. If UTC parameters have # also been requested we send them before the time (see later). If UTC parameters have not # been requested, we send time as the first packet if((do_settime == True) and (3 not in subtypes)): # If the user requested to set time on the device, but did not request # subtype 3 (Iono/UTC), we send the time here first time_success = client.send_time(sock=send_sock) sent_time = True # When we get the IONO/UTC parameters, there is a PRN field. However, # There is only one IONO/UTC parameter, so if we have more than one # file, only send the first one. sent_subtype3 = False # If only --send is present (no --get), scan output_dir and send all files for each subtype if do_send and not do_get: for st in subtypes: # Find all files for this subtype in output_dir pattern = os.path.join(output_dir, f"subtype_{st}_prn_*.json") files = sorted(glob.glob(pattern)) for file_path in files: if((st == 3) and (sent_subtype3 == True)): continue try: with open(file_path, 'r') as f: send_data_params = json.load(f) prn = send_data_params.get('sv_number', send_data_params.get('SV_Number', 0)) prn_send_success = client.send_data(st, send_data_params, sock=send_sock) send_results.append({ 'subtype': st, 'prn': prn, 'success': prn_send_success }) if prn_send_success: if(st == 3): sent_subtype3 = True success_count += 1 else: fail_count += 1 if((sent_time == False) and do_settime and (st == 3)): # If the user requested time sent to the device, we always send it # after the subtype 3. If no subtype 3 was requested, we'll have # already sent it time_success = client.send_time(sock=send_sock) send_results.append({ 'subtype': 0, # Time is subtype 0 'prn': 0, # No PRN for time 'success': time_success }) if time_success: success_count += 1 else: fail_count += 1 sent_time = True except Exception as e: fail_count += 1 continue # After sending all, skip the rest of the normal loop get_results = [] time_result = None # Close connections and return results if send_sock: try: send_sock.shutdown(socket.SHUT_RDWR) except: pass send_sock.close() traffic_file = None if client.traffic_log: traffic_file = os.path.join(client.output_dir, 'traffic.json') try: with open(traffic_file, 'w') as f: json.dump({ "summary": { "total_packets": len(client.traffic_log), "sent_packets": len([t for t in client.traffic_log if t["direction"] == "sent"]), "received_packets": len([t for t in client.traffic_log if t["direction"] == "received"]), "get_endpoint_packets": len([t for t in client.traffic_log if t["endpoint"] == "get"]), "send_endpoint_packets": len([t for t in client.traffic_log if t["endpoint"] == "send"]) }, "packets": client.traffic_log }, f, indent=2) except Exception as e: pass # Non-fatal error return { 'success': fail_count == 0, 'get_results': get_results, 'time_result': time_result, 'send_results': send_results, 'success_count': success_count, 'fail_count': fail_count, 'traffic_log': client.traffic_log, 'traffic_file': traffic_file } # To get here we are either doing do_get, or both do_get and do_send # Loop over subtypes and process each for current_subtype in subtypes: # Determine send_subtype for this iteration current_send_subtype = None if do_send: if send_subtype is not None: current_send_subtype = send_subtype elif do_get and current_subtype is not None: current_send_subtype = GET_TO_SEND_SUBTYPE_MAP.get(current_subtype) if current_send_subtype is None: if not multi_mode: raise ValueError(f"No automatic mapping from GET subtype {current_subtype} to SEND subtype. Provide send_subtype explicitly.") # else: skip this iteration (previously 'continue') # If not in a loop, just return or break out continue elif not do_get: current_send_subtype = current_subtype if current_subtype else subtypes[0] if subtypes else None # Determine PRN list for current subtype if str(prn).lower() == 'all': min_prn, max_prn = get_prn_range_for_subtype(current_subtype if do_get else None, current_send_subtype if do_send else None) current_prn_list = list(range(min_prn, max_prn + 1)) else: current_prn_list = prn_list # Process each PRN for current subtype for current_prn in current_prn_list: prn_get_result = None prn_send_success = False if((current_subtype == 3) and (sent_subtype3 == True)): continue if do_get: prn_get_result = client.get_data( subtype=current_subtype, sv_prn=current_prn, flags=flags, sat_type=sat_type, mode=mode, sock=get_sock ) if prn_get_result: get_results.append({ 'subtype': current_subtype, 'prn': current_prn, 'data': prn_get_result }) # Write to DAT file if requested if datfile and 'decoded' in prn_get_result: if current_subtype == 1: # GPS ephemeris client._write_gps_ephemeris_to_dat(current_prn, prn_get_result['decoded']) elif current_subtype == 11: # Galileo ephemeris client._write_galileo_ephemeris_to_dat(current_prn, prn_get_result['decoded']) else: if do_send: if multi_mode: fail_count += 1 continue else: raise RuntimeError("GET operation failed") elif not multi_mode: raise RuntimeError("GET operation failed") if do_send: # Determine data to send if data_params: send_data_params = data_params elif data_file: try: with open(data_file, 'r') as f: send_data_params = json.load(f) except Exception as e: if not multi_mode: raise ValueError(f"Error reading data file: {e}") else: fail_count += 1 continue elif prn_get_result and do_get: send_data_params = prn_get_result.get('decoded', {}) else: if not multi_mode: raise ValueError("No data available for SEND operation. Provide data_params, data_file, or capture data with do_get first.") else: fail_count += 1 continue prn_send_success = client.send_data(current_send_subtype, send_data_params, sock=send_sock) send_results.append({ 'subtype': current_send_subtype, 'prn': current_prn, 'success': prn_send_success }) if not prn_send_success: if not multi_mode: raise RuntimeError("SEND operation failed") if((sent_time == False) and do_settime and (current_subtype == 3)): # If the user requested time sent to the device, we always send it # after the subtype 3. If no subtype 3 was requested, we'll have # already sent it time_success = client.send_time(sock=send_sock) send_results.append({ 'subtype': 0, # Time is subtype 0 'prn': 0, # There isn't a PRN for time 'success': time_success }) if time_success: success_count += 1 else: fail_count += 1 sent_time = True if (not do_get or prn_get_result) and (not do_send or prn_send_success): if(current_subtype == 3): sent_subtype3 = True success_count += 1 else: fail_count += 1 finally: # Close persistent connections if get_sock: try: get_sock.shutdown(socket.SHUT_RDWR) except: pass get_sock.close() if send_sock: try: send_sock.shutdown(socket.SHUT_RDWR) except: pass send_sock.close() # Always write traffic log, even if empty or only time-transfer operations were requested traffic_file = os.path.join(client.output_dir, 'traffic.json') try: os.makedirs(client.output_dir, exist_ok=True) with open(traffic_file, 'w') as f: json.dump({ "summary": { "total_packets": len(client.traffic_log), "sent_packets": len([t for t in client.traffic_log if t.get("direction") == "sent"]), "received_packets": len([t for t in client.traffic_log if t.get("direction") == "received"]), "get_endpoint_packets": len([t for t in client.traffic_log if t.get("endpoint") == "get"]), "send_endpoint_packets": len([t for t in client.traffic_log if t.get("endpoint") == "send"]) }, "packets": client.traffic_log }, f, indent=2) except Exception as e: print(f"Warning: Could not write traffic.json: {e}") return { 'success': fail_count == 0, 'get_results': get_results, 'time_result': time_result, 'send_results': send_results, 'success_count': success_count, 'fail_count': fail_count, 'traffic_log': client.traffic_log, 'traffic_file': traffic_file } def main(): """Main function with command-line interface""" parser = argparse.ArgumentParser( description='DCOL Protocol Client - Get and Send satellite data', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Get GPS ephemeris for satellite 5 %(prog)s --get-server 192.168.1.100 --get-port 5000 --get --subtype 1 --prn 5 # Get GPS ephemeris for all GPS satellites (1-32) %(prog)s --get-server 192.168.1.100 --get-port 5000 --get --subtype 1 --prn All # Send time and date %(prog)s --send-server 192.168.1.100 --send-port 5000 --send --send-subtype 0 --data '{"utc_week": 2234, "utc_seconds": 432000}' # Get GPS ephemeris and send to another server (auto-maps subtype 1 -> 2) %(prog)s --get-server 10.1.150.XXX --get-port 6001 --send-server 192.168.1.200 --send-port 5000 --get --send --subtype 1 --prn 5 # Get all Galileo almanacs and forward to another server %(prog)s --get-server 10.1.150.XXX --get-port 6001 --send-server 192.168.1.200 --send-port 5000 --get --send --subtype 12 --prn All # Get and send with explicit send subtype override %(prog)s --get-server 192.168.1.100 --get-port 5000 --send-server 192.168.1.100 --send-port 5000 --get --send --subtype 1 --prn 3 --send-subtype 2 # Get GLONASS ephemeris and forward (auto-maps subtype 9 -> 7) %(prog)s --get-server 10.1.150.XXX --get-port 6001 --send-server 192.168.1.200 --send-port 5000 --get --send --subtype 9 --prn 1 # Get multiple subtypes (GPS ephemeris, GPS almanac, Galileo ephemeris, Galileo almanac) %(prog)s --get-server 192.168.1.100 --get-port 5000 --get --subtype 1,2,11,12 --prn 5 """) # Server configuration parser.add_argument('--get-server', help='Server IP address for GET operations') parser.add_argument('--get-port', type=int, help='Server port number for GET operations') parser.add_argument('--send-server', help='Server IP address for SEND operations') parser.add_argument('--send-port', type=int, help='Server port number for SEND operations') # Operation mode (now both can be specified) parser.add_argument('--get', action='store_true', help='Get data from server (54h GETSVDATA)') parser.add_argument('--settime', action='store_true', help='Sets the time on the receiver') parser.add_argument('--send', action='store_true', help='Send data to server (A9h)') # Common parameters parser.add_argument('--output-dir', default='dcol_data', help='Output directory for saved data (default: dcol_data)') parser.add_argument('--debug', action='store_true', help='Enable debug output (show hex bytes for each parameter)') parser.add_argument('--radians', action='store_true', help='Convert semi-circle parameters to radians (multiply by PI)') # Get-specific parameters parser.add_argument('--get-subtype', type=int, help='Subtype value for GET operation') parser.add_argument('--prn', default=0, help='Satellite PRN number or "All" to loop through all PRNs for the constellation') parser.add_argument('--flags', type=lambda x: int(x, 0), default=0, help='Flags field (for get, can be hex like 0x01)') parser.add_argument('--sat-type', type=int, default=0, help='Satellite type (for get subtype 20)') parser.add_argument('--mode', type=int, default=0, help='Mode (for get subtype 20)') # Placeholder flag for future GETTIME support (currently unused) #parser.add_argument('--uptime', action='store_true', help='Reserved for future GETTIME support (no effect at present)') # DAT file output parser.add_argument('--datfile', help='Path to Trimble DAT file for writing ephemeris records (binary format)') # Send-specific parameters parser.add_argument('--send-subtype', type=int, help='Subtype value for SEND operation (overrides automatic mapping)') parser.add_argument('--data', help='JSON string with data parameters (for send)') parser.add_argument('--data-file', help='JSON file with data parameters (for send)') parser.add_argument('--pos', help='Position as latitude,longitude (e.g., 37.7749,-122.4194). Sent as LAST_KNOWN_POS (subtype 5) before other data when --send is used') # Subtype parameter - when both --get and --send are specified, this refers to GET subtype parser.add_argument('--subtype', help='Subtype value(s) - single integer or comma-separated list (e.g. 1,7,11,12) for GET operation; SEND subtype auto-mapped if both operations specified') args = parser.parse_args() # Parse subtype(s) - can be a single value or comma-separated list subtype_arg = args.get_subtype if args.get_subtype is not None else args.subtype if subtype_arg is not None: if isinstance(subtype_arg, str): # Parse comma-separated list try: subtype_list = [int(s.strip()) for s in subtype_arg.split(',')] except ValueError: parser.error(f"Invalid subtype format: {subtype_arg}. Must be integer or comma-separated integers (e.g. 1,7,11,12)") else: # Single integer from --get-subtype subtype_list = [subtype_arg] # Validate all GET subtypes if args.get: for st in subtype_list: try: validate_subtype(st, is_get=True) except ValueError as e: parser.error(str(e)) else: subtype_list = None # Validate SEND subtype if provided if args.send_subtype is not None: try: validate_subtype(args.send_subtype, is_get=False) except ValueError as e: parser.error(str(e)) # Parse and validate position if provided position_params = None if args.pos: try: parts = args.pos.split(',') if len(parts) != 2: parser.error(f"Invalid position format: {args.pos}. Must be latitude,longitude (e.g., 37.7749,-122.4194)") latitude = float(parts[0].strip()) longitude = float(parts[1].strip()) # Validate latitude (-90 to +90) if latitude < -90 or latitude > 90: parser.error(f"Invalid latitude: {latitude}. Must be between -90 and +90") # Validate longitude (-360 to +360) if longitude < -360 or longitude > 360: parser.error(f"Invalid longitude: {longitude}. Must be between -360 and +360") # Normalize longitude to -180 to +180 range while longitude > 180: longitude -= 360 while longitude < -180: longitude += 360 position_params = { 'reserved': 0, 'latitude': latitude, 'longitude': longitude, 'height': 0.0 # Default height to 0 } if args.debug: print(f"Position parsed: lat={latitude}, lon={longitude}") except ValueError as e: parser.error(f"Invalid position values: {args.pos}. Must be numeric values") # Parse data if provided data_params = None if args.data: try: data_params = json.loads(args.data) except json.JSONDecodeError as e: parser.error(f"Invalid JSON in --data: {e}") # Determine if in multi mode (for verbose output control) prn_list_mode = str(args.prn).lower() == 'all' multi_mode = prn_list_mode or (subtype_list and len(subtype_list) > 1) # Print initial status messages if prn_list_mode and subtype_list: get_subtype = subtype_list[0] min_prn, max_prn = get_prn_range_for_subtype(get_subtype if args.get else None, args.send_subtype if args.send else None) print(f"PRN 'All' mode: Will process PRNs {min_prn} to {max_prn} ({max_prn - min_prn + 1} satellites)") if multi_mode: if args.get and args.get_server and args.get_port: print(f"Opened persistent GET connection to {args.get_server}:{args.get_port}") if args.send and args.send_server and args.send_port: print(f"Opened persistent SEND connection to {args.send_server}:{args.send_port}") # Print subtype processing headers if subtype_list and len(subtype_list) > 1: for st in subtype_list: print(f"\n{'='*60}") print(f"Processing Subtype {st}") print(f"{'='*60}") # Show auto-mapping if applicable if args.get and args.send: send_st = GET_TO_SEND_SUBTYPE_MAP.get(st) if args.send_subtype is None else args.send_subtype if send_st is not None and args.send_subtype is None: print(f"Auto-mapping: GET subtype {st} -> SEND subtype {send_st}") if prn_list_mode: min_prn, max_prn = get_prn_range_for_subtype(st if args.get else None, args.send_subtype if args.send else None) print(f"PRN 'All' mode for subtype {st}: Will process PRNs {min_prn} to {max_prn} ({max_prn - min_prn + 1} satellites)") elif subtype_list and len(subtype_list) == 1 and args.get and args.send: # Single subtype but both GET and SEND send_st = GET_TO_SEND_SUBTYPE_MAP.get(subtype_list[0]) if args.send_subtype is None else args.send_subtype if send_st is not None and args.send_subtype is None: print(f"Auto-mapping: GET subtype {subtype_list[0]} -> SEND subtype {send_st}") try: # Call the core function result = run_dcol_operations( get_server_ip=args.get_server, get_port=args.get_port, send_server_ip=args.send_server, send_port=args.send_port, do_get=args.get, do_settime=args.settime, do_send=args.send, subtypes=subtype_list, prn=args.prn, flags=args.flags, sat_type=args.sat_type, mode=args.mode, send_subtype=args.send_subtype, data_params=data_params, data_file=args.data_file, output_dir=args.output_dir, debug=args.debug, radians=args.radians, datfile=args.datfile, position_params=position_params ) # Print results based on mode if not multi_mode: # Single operation mode - print detailed results if result['get_results']: print("\nReceived data:") # Create a temporary client just for formatting temp_client = DCOLClient( get_server_ip=args.get_server, get_port=args.get_port, debug=args.debug, radians=args.radians ) temp_client._print_formatted_data(result['get_results'][0]['data']) if result['time_result']: print("\nGPS Time:") temp_client = DCOLClient( get_server_ip=args.get_server, get_port=args.get_port, debug=args.debug, radians=args.radians ) temp_client._print_formatted_data(result['time_result']) if result['send_results']: if result['send_results'][0]['success']: print("Data sent successfully") else: print("Failed to send data") # Print connection close messages if multi_mode: if args.get: print(f"\nClosed GET connection") if args.send: print(f"Closed SEND connection") # Print traffic log location if result['traffic_file']: print(f"\nTraffic log saved to: {result['traffic_file']}") # Print summary for multi mode if multi_mode: # Build separate summaries for GET and SEND get_stats = {} for entry in result.get('get_results', []): st = entry.get('subtype') if st is None: continue if st not in get_stats: get_stats[st] = {'success': 0, 'fail': 0, 'total': 0} if entry.get('data'): get_stats[st]['success'] += 1 else: get_stats[st]['fail'] += 1 get_stats[st]['total'] += 1 send_stats = {} for entry in result.get('send_results', []): st = entry.get('subtype') if st is None: continue if st not in send_stats: send_stats[st] = {'success': 0, 'fail': 0, 'total': 0} if entry.get('success'): send_stats[st]['success'] += 1 else: send_stats[st]['fail'] += 1 send_stats[st]['total'] += 1 print(f"\n{'='*60}") print("GET Summary by Subtype:") for st, stats in get_stats.items(): print(f"Subtype {st}: {stats['success']} successful, {stats['fail']} failed out of {stats['total']} operations") print(f"{'='*60}") print("SEND Summary by Subtype:") for st, stats in send_stats.items(): print(f"Subtype {st}: {stats['success']} successful, {stats['fail']} failed out of {stats['total']} operations") print(f"{'='*60}") # Exit with appropriate code sys.exit(0 if result['success'] else 1) except (ValueError, RuntimeError, ConnectionError) as e: print(f"Error: {e}") sys.exit(1) except Exception as e: print(f"Unexpected error: {e}") sys.exit(1) if __name__ == '__main__': main()