""" 2020 - 2025 Copyright Trimble Inc. Class, ParseRT27(), for decoding an RT27 stream. The caller uses ParseRT27.process_data() to pass binary data into the decoder, it will return a code(s) indicating whether there is new decoded data. The caller can access the decoded data from the object using ParseRT27.get_decoded_obs() - list of decoded observable records of type RT27Obs() ParseRT27.get_decoded_pos() - list of decoded position records of type RT27Pos() [not decoded yet] ParseRT27.get_decoded_fft() - list of decoded FFT records of type RT27Fft(). ParseRT27.get_legacy_output() - Legacy interface, limited to one record per call to process_data() There legacy out provides two decoded structures "obsEpoch", this provides data in a Trimble format", e.g. it retains meas flags. The other format is obsGenericEpoch", the idea is this will abstract from the receiver and so we can write similar decoders for other manf. for comparison testing. For example usage, see the the __main__ functionality to decode RT27 from a file or from an active IP stream. """ import sys import socket import struct from enum import Enum ########################################################################## # RT27 decoding utilities class class RT27Decode(): """ Utility methods for decoding RT27 data All are static methods """ @staticmethod def decode_u16(data, start): """ Decode a U16 from data, begining a index 'start' """ return data[start+1] + (data[start]<<8) @staticmethod def decode_flt(data, start): """ Decode a float/FLT from data, begining a index 'start' """ return struct.unpack('>f', bytes(data[start:start+4]))[0] @staticmethod def decode_s16(data, start): """ Decode a S16 from data, begining a index 'start' """ tmp = RT27Decode.decode_u16(data, start) return RT27Decode.twos_complement(tmp, 16) @staticmethod def decode_u24(data, start): """ Decode a U24 from data, begining a index 'start' """ return data[start+2] + (data[start+1]<<8) + (data[start]<<16) @staticmethod def decode_s24(data, start): """ Decode a S24 from data, begining a index 'start' """ tmp = RT27Decode.decode_u24(data, start) return RT27Decode.twos_complement(tmp, 24) @staticmethod def decode_u32(data, start): """ Decode a U32 from data, begining a index 'start' """ return data[start+3] + (data[start+2]<<8) + (data[start+1]<<16) + (data[start]<<24) @staticmethod def decode_s32(data, start): """ Decode a S32 from data, begining a index 'start' """ tmp = RT27Decode.decode_u32(data, start) return RT27Decode.twos_complement(tmp, 32) @staticmethod def decode_u48(data, start): """ Decode a U48 from data, begining a index 'start' """ return (RT27Decode.decode_u24(data, start)<<24) + RT27Decode.decode_u24(data, start+3) @staticmethod def decode_s48(data, start): """ Decode a S48 from data, begining a index 'start' """ tmp = RT27Decode.decode_u48(data, start) return RT27Decode.twos_complement(tmp, 48) @staticmethod def twos_complement(value, n_bits): """ Two's complement calculation, unsigned -> signed """ sign_mask = 1 << (n_bits-1) if value & sign_mask: value -= 2 * sign_mask return value @staticmethod def get_next_block_idx(data, idx): """ Get the index of the next data block based on the current block starting at index 'idx' Bounds checks is ensure there is enough bytes in 'data' Returns -1 if there is insufficient bytes """ if idx < 0 or idx >= len(data): return -1 idx += data[idx] if idx > len(data): return -1 return idx @staticmethod def calc_block_start_idx(data, block): """ Calculate the index into 'data' for the requested block number Bounds checks is ensure there is enough bytes in 'data' Returns -1 if there is insufficient bytes """ s_idx = 0 e_idx = 0 while block >= 0: s_idx = e_idx e_idx = RT27Decode.get_next_block_idx(data, s_idx) block -= 1 return s_idx ########################################################################## # RT27 observables class class RT27Obs(): """ RT27 observables data storage class Populates the class based on the contents of 'data' if not None """ def __init__(self, data=None): self.hdr = RT27Obs.RT27Hdr(data) self.clkblk = RT27Obs.RT27ClkBlk(data) self.meas = RT27Obs.RT27Meas.generate_meas_list(data, self.hdr.num_svs) class RT27Hdr(): """ RT27 observables header class, contains info. common to all the following measurements, such as time of week Populates the class based on the contents of 'data' if not None """ def __init__(self, data=None): self.week = None self.sec = None self.bias = None self.num_svs = None self.populate_self(data) def populate_self(self, data): """ Populate the class based on the contents of 'data' """ if data is None or len(data) == 0: return s_idx = RT27Decode.calc_block_start_idx(data, 0) if s_idx < 0: return self.week = RT27Decode.decode_u16(data, s_idx+1) self.sec = RT27Decode.decode_u32(data, s_idx+3) scale = float(1 << 19) self.bias = RT27Decode.decode_s24(data, s_idx+7) / scale self.num_svs = data[s_idx+10] class RT27ClkBlk(): """ RT27 observable clock bias block class Not implemented yet """ def __init__(self, data=None): self.populate_self(data) def populate_self(self, data): """ Populate the class based on the contents of 'data' Not implemented yet """ if data is None or len(data) == 0: return s_idx = RT27Decode.calc_block_start_idx(data, 1) if s_idx < 0: return class RT27Meas(): """ RT27 observables class, contains tracking info. for a single signal from a single SV Populates the class based on the contents of 'data' if not None If 'data' is not None then the initialiser needs to know - the index into 'data' for the SV header block - the index into 'data' for the the measurement block - pseudo_base, None if this is the first measurement block for this SV otherwise equal to 'pseudo' from the first measurement block """ def __init__(self, data=None, sv_idx=-1, meas_idx=-1, pseudo_base=None): self.chan = None self.sv_id = None self.sat_type = None self.ant_num = None self.elev = None self.azi = None self.sv_flags = [None] * 2 self.freq = None self.track = None self.cno = None self.pseudo = None self.phase = None self.slip_count = None self.meas_flags = [0] * 4 self.doppler = None self.populate_self(data, sv_idx, meas_idx, pseudo_base=pseudo_base) def populate_self(self, data, sv_idx, meas_idx, pseudo_base=None): """ Populate the class based on the contents of 'data' See the class comments for other input parameters """ if data is None or len(data) == 0: print('Data length error 1') return len_data = len(data) if ((sv_idx>=0 and len_data < sv_idx) or RT27Decode.get_next_block_idx(data, sv_idx) < 0): print('Data length error 2a',len_data,sv_idx) return if ((meas_idx>=0 and len_data < meas_idx) or RT27Decode.get_next_block_idx(data, meas_idx) < 0): print('Data length error 3a',meas_idx,len_data) return self.populate_sv(data, sv_idx) self.populate_meas(data, meas_idx, pseudo_base=pseudo_base) def populate_sv(self, data, sv_idx): """ Populate the SV header based on the contents of 'data' See the class comments for other input parameters """ if len(data) < sv_idx or RT27Decode.get_next_block_idx(data, sv_idx) < 0: print('Data length error 2b') return self.sv_id = data[sv_idx+1] self.sat_type = data[sv_idx+2] & 0x3f self.ant_num = data[sv_idx+2] >> 6 self.chan = data[sv_idx+3] self.elev = RT27Decode.twos_complement(data[sv_idx+5], 8) self.azi = data[sv_idx+6] self.sv_flags[0] = data[sv_idx+7] self.sv_flags[1] = 0 if self.sv_flags[0] & 0x80: self.sv_flags[1] = data[sv_idx+8] def populate_meas(self, data, meas_idx, pseudo_base=None): """ Populate the measurements based on the contents of 'data' See the class comments for other input parameters """ if len(data) < meas_idx or RT27Decode.get_next_block_idx(data, meas_idx) < 0: print('Data length error 3b') return self.freq = data[meas_idx+1] self.track = data[meas_idx+2] self.cno = RT27Decode.decode_u16(data, meas_idx+3) / 10.0 # The pseudorange is 4-bit for block zero and a 2-bit difference # in other blocks # It is constructed later as it is dependent on meas_flags # ppr - post pseudorane index ppr_idx = meas_idx + 7 if pseudo_base is None: ppr_idx = meas_idx + 9 scale = float(1 << 15) self.phase = RT27Decode.decode_s48(data, ppr_idx) / scale self.slip_count = data[ppr_idx+6] # Optional flags, MSB signs another byte follows # Assume four worst case for now self.meas_flags = [0] * 4 flag_idx = 0 self.meas_flags[flag_idx] = data[ppr_idx+7+flag_idx] while self.meas_flags[flag_idx] & 0x80 > 0: flag_idx += 1 self.meas_flags[flag_idx] = data[ppr_idx+7+flag_idx] pfl_idx = ppr_idx + 8 + flag_idx pdp_idx = pfl_idx if self.meas_flags[0] & 0x4 > 0: self.doppler = RT27Decode.decode_s24(data, pfl_idx) / 256.0 # Convert to Hz pdp_idx += 3 else: self.doppler = 0.0 # Finally construct the pseudorange if pseudo_base is None: scale = 128.0 if self.sat_type in [1, 4]: scale = 64.0 self.pseudo = RT27Decode.decode_u32(data, meas_idx+5) / scale if self.meas_flags[1] & 0x2 > 0: self.pseudo += float(1 << 25) - 1.0 else: delta_pr = RT27Decode.decode_s16(data, meas_idx+5) if self.meas_flags[1] & 0x2 > 0: delta_3b = RT27Decode.decode_u16(data, meas_idx+5) + (data[pdp_idx] << 16) delta_pr = RT27Decode.twos_complement(delta_3b, 24) scale = 256.0 self.pseudo = pseudo_base + delta_pr / scale @staticmethod def _get_num_sv_meas(data, idx): """ Get the number of measurements from the SV measurement header block Bounds checks is ensure there is enough bytes in 'data' Returns -1 if there is insufficient bytes """ if idx+4 >= len(data) or idx < 0: return -1 return data[idx + 4] @staticmethod def generate_meas_list(data, num_svs): """ Static method Call to generate a list of RT27Meas blocks based on the contents of 'data' """ if data is None or len(data) == 0 or num_svs == 0: return [] meas_list = [] # Get one block early, it'll be increased in the for loop sv_idx = RT27Decode.calc_block_start_idx(data, 2-1) meas_idx = sv_idx for _ in range(num_svs): sv_idx = RT27Decode.get_next_block_idx(data, meas_idx) meas_idx = sv_idx num_meas = RT27Obs.RT27Meas._get_num_sv_meas(data, sv_idx) pseudo_base = None # Signifies this block is the pseudorange base for _ in range(num_meas): meas_idx = RT27Decode.get_next_block_idx(data, meas_idx) rt27_meas = RT27Obs.RT27Meas(data, sv_idx, meas_idx, pseudo_base=pseudo_base) meas_list.append(rt27_meas) if pseudo_base is None: # Now have a real pseudorange base pseudo_base = rt27_meas.pseudo return meas_list ########################################################################## # RT27 observables class class RT27Pos(): """ RT27 observables data storage class Not implemented yet """ def __init__(self): pass class RT27OneFft(): def __init__(self): self.ant = "" self.miti = "" self.freq = "" self.mode = "" self.points = [] class RT27MetaFft(): def __init__(self): self.fft_base = [] self.gain = [] self.RSSI = [] self.quality = [] self.commonRSSI = [] self.post_quality = [] class RT27Fft(): """ RT27 FFT data storage class Populates the class based on the contents of 'data' """ def __init__(self,data): """Initialize structure and fill in data""" self.msecs = None self.fft = [] self.meta = RT27MetaFft() self.populate_self(data) def populate_self(self, data): """Parse RT27 FFT 'data' into local structure""" if data is None or len(data) == 0: return s_idx = RT27Decode.calc_block_start_idx(data, 0) if s_idx < 0: return self.msecs = RT27Decode.decode_u32(data, s_idx) s_idx += 4 n_bands = data[s_idx] s_idx += 1 n_freqs = data[s_idx] s_idx += 1 config = RT27Decode.decode_u32(data, s_idx) s_idx += 4 if config&(1<<31): config2 = RT27Decode.decode_u32(data, s_idx) s_idx += 4 if (config&3) == 2: all_modes = ['avg','max'] elif (config&3) == 0: all_modes = ['avg'] else: all_modes = ['max'] miti = (config>>2)&3 if miti == 0: all_miti = ['pre'] elif miti == 1: all_miti = ['post'] else: all_miti = ['pre','post'] ant = (config>>4)&3 if ant == 0: all_ant = ['0'] elif ant == 1: all_ant = ['1'] else: all_ant = ['0','1'] add_tot_gain = ((config>>6)&1) != 0 add_rssi = ((config>>7)&1) != 0 add_qual = ((config>>8)&1) != 0 add_FE_rssi = ((config>>26)&1) != 0 add_post_qual = ((config>>27)&1) != 0 prefix = '' spacing = 1 << ((config>>9)&7) raw_points = (config>>12)&0x7 if raw_points == 0: prefix = 'all' n_points = 2048//spacing else: n_points = 2 * (2**raw_points) + 1 known_freqs = [] if config & (1<<15): known_freqs.append(prefix+'L1') if config & (1<<16): known_freqs.append(prefix+'L2') if config & (1<<17): known_freqs.append(prefix+'L5') if config & (1<<18): known_freqs.append(prefix+'E6') if config & (1<<19): known_freqs.append(prefix+'B1') if config & (1<<20): known_freqs.append(prefix+'S1') known_bands = list(known_freqs) for i in range(len(known_bands),n_bands): known_bands.append( f"unknown {i}" ) if config & (1<<21): known_freqs.extend(['G1(%d)'%x for x in range(-7,7)]) if config & (1<<22): known_freqs.extend(['G2(%d)'%x for x in range(-7,7)]) if config & (1<<23): known_freqs.append('G3') if config & (1<<24): known_freqs.append('E5B') if config & (1<<25): known_freqs.append('B3') for i in range(len(known_freqs),n_freqs): known_freqs.append( f"unknown {i}" ) #print(n_freqs,known_freqs,n_bands,all_ant,all_modes,all_miti,n_points,len(data)) for freq in range(n_freqs): for ant in range(len(all_ant)): for miti in range(len(all_miti)): for mode in range(len(all_modes)): freq_name = known_freqs[freq] curr_fft = RT27OneFft() curr_fft.miti = all_miti[miti] curr_fft.ant = all_ant[ant] curr_fft.freq = freq_name curr_fft.mode = all_modes[mode] curr_fft.points = [data[s_idx+i]/4.0 + 10 for i in range(n_points)] self.fft.append( curr_fft ) s_idx += n_points if prefix == 'all': for band in range(n_bands): band_name = known_bands[band] freq = RT27Decode.decode_flt(data, s_idx) s_idx += 4 self.meta.fft_base.append( [band_name, freq] ) if add_tot_gain: for band in range(n_bands): band_name = known_bands[band] for ant in range(len(all_ant)): gain = RT27Decode.decode_s16(data,s_idx) s_idx += 2 self.meta.gain.append( [band_name, all_ant[ant], gain] ) if add_rssi: for band in range(n_bands): band_name = known_bands[band] for ant in range(len(all_ant)): rssi = RT27Decode.twos_complement(data[s_idx],8) s_idx += 1 self.meta.RSSI.append( [band_name, all_ant[ant], rssi] ) if add_qual: for band in range(n_bands): band_name = known_bands[band] for ant in range(len(all_ant)): qual = data[s_idx] s_idx += 1 self.meta.quality.append( [band_name, all_ant[ant], qual] ) if add_FE_rssi: FE_len = data[s_idx] s_idx += 1 for idx in range(FE_len): common_name = data[s_idx]; s_idx += 1 antenna = data[s_idx]; s_idx += 1 rssi = RT27Decode.twos_complement(data[s_idx],8); s_idx += 1 self.meta.commonRSSI.append( [common_name, antenna, rssi] ) if add_post_qual: for band in range(n_bands): band_name = known_bands[band] for ant in range(len(all_ant)): qual = data[s_idx] s_idx += 1 self.meta.post_quality.append( [band_name, all_ant[ant], qual] ) if s_idx != len(data): print("************************************************************") print(f"Warning: FFT undecoded data... proc={s_idx} tot={len(data)}") print("************************************************************") ########################################################################## # Utilities for RT27 stream parsing class RT27Utils(): """ Utility methods for parsing RT27 stream All are static methods """ @staticmethod def check_packet_integrity(packet, length): """ Check the packet integrity """ if packet[0] != 0x02 or packet[length-1] != 0x03: return False chk_sum = 0 for idx in range(1, length-2): chk_sum += packet[idx] chk_sum &= 0xff return chk_sum == packet[length-2] @staticmethod def packet_rt27_marker(packet): """ Check for the RT27 data marker """ return packet[2] == 0x57 @staticmethod def packet_length(packet): """ Return the RT27 packet length """ return packet[3] + 6 @staticmethod def packet_record(packet): """ Return the RT27 packet record type """ return packet[4] ########################################################################## # RT27 parser class class ParseRT27(): """ RT27 stream parser class """ def __init__(self, show_warnings=True, show_info=False): self._data = [] # All data self._payload_data = [] # Concat. of multi-packet data self._decoded_obs = [] # List of decode observable records self._decoded_pos = [] # List of decode position records self._decoded_fft = [] # List of decode FFT records self._last_sequence = None self._show_warnings = show_warnings self._show_info = show_info self._last_page = 0 self._page_wrap = 0 self._have_page_1 = False # Used to construct the legacy outputs self.last_obs_hdr = None def _reset_concat_data(self, page): """ Reset the data concatenation Either found a decoding error, such as packets out of sequence Or the full packet is complete and now decoded """ self._payload_data = [] self._last_page = page self._page_wrap = 0 self._have_page_1 = (page == 1) def _reset_decoded_lists(self): """ Reset the lists of decoded records Call when a new decoding is requested """ self._decoded_obs = [] self._decoded_pos = [] self._decoded_fft = [] def _append_payload(self, packet): """ Got a good packet so append the payload to the existing bytes for multi-page payloads """ # Remove STX and packet header info. # Remove CSum and ETX self._payload_data += packet[8:-2] def _append_decoded_obs(self, new_obs): """ Append a decoded observable to the list """ self._decoded_obs.append(new_obs) self.last_obs_hdr = new_obs.hdr def _append_decoded_pos(self, new_pos): """ Append a decoded position to the list """ self._decoded_pos.append(new_pos) def _append_decoded_fft(self, new_fft): """ Append a decoded FFT to the list """ self._decoded_fft.append(new_fft) def _print_warnings(self, message): """ Print a decoder warning level message, if enabled """ if self._show_warnings: print('WARNING: '+message) def _print_info(self, message): """ Print a decoder info. level message, if enabled """ if self._show_info: print('INFO: '+message) class Records(): """ RT27 record types """ def __init__(self): self.obs = 6 self.pos = 7 self.fft = 18 class CheckState(Enum): """ Enumerate the decoded packet state """ GOOD_PACKET_COMPLETE = 0 GOOD_PACKET_INCOMPLETE = 1 BAD_PACKET = 2 NON_RT27_PACKET = 3 INSUFFICIENT_DATA = 4 def _check_inc_sequence(self, sequence, this_page): """ Packet decoding checker Ensure that the sequence number increments across complete payloads """ last_sequence = self._last_sequence self._last_sequence = sequence self._last_page = this_page if last_sequence is not None: last_sequence = (last_sequence + 1) & 0xff if sequence != last_sequence: self._print_warnings('Data out-of-sequence detected %d %d'%(sequence,last_sequence)) self._reset_concat_data(this_page) def _check_pages(self, sequence, this_page): """ Packet decoding checker For multi-page packets, ensure that the page number increments and that the sequence does not across pages """ if sequence != self._last_sequence: self._print_warnings('Page sequence changed') self._reset_concat_data(this_page) return last_page = self._last_page + 1 self._last_page = this_page if this_page != last_page: self._print_warnings(f'Pages out-of-sequence detected {this_page} {last_page}') self._reset_concat_data(this_page) def _packet_pages(self,packet): """ Decode and return the packet page information """ n_extra_15 = (packet[7]>>2)&0x1f page_info = packet[5] this_page = page_info >> 4 num_pages = page_info & 0xf num_pages += n_extra_15*15 sequence = packet[6] if sequence != self._last_sequence: self._page_wrap = 0 elif (self._last_page%15)==0 and this_page==1: if self._page_wrap<=(n_extra_15-1)*15: self._page_wrap += 15 return (this_page+self._page_wrap, num_pages, sequence) def _check_packet(self): """ Packet decoding checker """ len_data = len(self._data) if len_data < 8: self._print_info('Insufficient data 1') return (ParseRT27.CheckState.INSUFFICIENT_DATA, 0) packet_length = RT27Utils.packet_length(self._data) if len_data < packet_length: self._print_info('Insufficient data 2') return (ParseRT27.CheckState.INSUFFICIENT_DATA, 0) if not RT27Utils.check_packet_integrity(self._data, packet_length): self._print_info('Bad packet framing or checksum') return (ParseRT27.CheckState.BAD_PACKET, 1) if not RT27Utils.packet_rt27_marker(self._data): self._print_info('Non-RT27 packet') return (ParseRT27.CheckState.NON_RT27_PACKET, packet_length) (this_page, num_pages, sequence) = self._packet_pages(self._data) if this_page == 1: self._check_inc_sequence(sequence, this_page) else: self._check_pages(sequence, this_page) packet = self._data[0:packet_length] self._append_payload(packet) state = ParseRT27.CheckState.GOOD_PACKET_INCOMPLETE info_str = 'incomplete' if this_page == num_pages and self._have_page_1: state = ParseRT27.CheckState.GOOD_PACKET_COMPLETE info_str = 'complete %d'%num_pages self._print_info('Good packet '+info_str) return (state, packet_length) def _process_packet(self, packet): """ A good, complete packet was found Decode the record it contains """ record = RT27Utils.packet_record(packet) if record == ParseRT27.Records().obs: new_obs = RT27Obs(self._payload_data) self._append_decoded_obs(new_obs) elif record == ParseRT27.Records().pos: new_pos = RT27Pos() self._append_decoded_pos(new_pos) elif record == ParseRT27.Records().fft: new_fft = RT27Fft(self._payload_data) self._append_decoded_fft(new_fft) else: self._print_info('Skipped RT27 record type '+str(record)) def process_data(self, data): """ Externally callable function Add 'data' to the list of bytes to decode Searches for the next valid packet and decodes any records found """ # Append new data bytes self._data += data self._reset_decoded_lists() # There may be multiple RT27 packets in the buffer # Loop around pushing each packet into decoder while True: (state, skip_bytes) = self._check_packet() if state == ParseRT27.CheckState.INSUFFICIENT_DATA: break if state == ParseRT27.CheckState.GOOD_PACKET_COMPLETE: self._process_packet(self._data[0:skip_bytes]) self._data = self._data[skip_bytes:] self._print_info('Num. decoded obs: '+str(len(self._decoded_obs))) self._print_info('Num. decoded pos: '+str(len(self._decoded_pos))) self._print_info('Num. decoded FFT: '+str(len(self._decoded_fft))) def get_decoded_obs(self): """ Get a list of the decoded observale records for the last processing call Empty list returned if none decoded """ return self._decoded_obs def get_decoded_fft(self): """ Get a list of the decoded FFT records for the last processing call Empty list returned if none decoded """ return self._decoded_fft def get_decoded_pos(self): """ Get a list of the decoded position records for the last processing call Empty list returned if none decoded """ return self._decoded_pos def get_legacy_output(self): """ Get the legacy outputs - RT27.ret - RT27.obsEpoch (dictionary) - RT27.week / secs / bias / numSVs """ dcd_obs = self._decoded_obs dcd_pos = self._decoded_pos week = 0 secs = 0 bias = 0 num_svs = 0 if self.last_obs_hdr is not None: week = self.last_obs_hdr.week secs = self.last_obs_hdr.sec bias = self.last_obs_hdr.bias num_svs = self.last_obs_hdr.num_svs legacy_output = ParseRT27Legacy(dcd_obs, dcd_pos, week, secs, bias, num_svs) return legacy_output def get_data_length(self): """ Externally callable function - legacy compatabilty Return the number of bytes remaining in the decoder """ return len(self._data) ########################################################################## class ParseRT27Legacy(): """ Legacy output class """ def __init__(self, rt27_obs, rt27_pos, week, secs, bias, num_svs): self.week = week self.secs = secs self.bias = bias self.numSVs = num_svs self.obsEpoch = [] self.obsGenericEpoch = [] if len(rt27_obs) > 0: for lgcy_meas in rt27_obs[-1].meas: (tmp1, tmp2) = self._construct_epochs(lgcy_meas) self.obsEpoch.append(tmp1) self.obsGenericEpoch.append(tmp2) ret_type = ParseRT27Legacy.ReturnType().dNoData if len(rt27_obs) > 0: ret_type = ret_type + ParseRT27Legacy.ReturnType().dObs if len(rt27_pos) > 0: ret_type = ret_type + ParseRT27Legacy.ReturnType().dPVT self.ret = ret_type @staticmethod def _construct_epochs(rt27_meas): """ Contruct the obsEpoch and obsGenericEpoch outputs """ obs_trmb = {} obs_gen = {} obs_trmb['chan'] = rt27_meas.chan obs_trmb['svID'] = obs_gen['svID'] = rt27_meas.sv_id obs_trmb['satType'] = obs_gen['satType'] = rt27_meas.sat_type obs_trmb['antNum'] = obs_gen['antNum'] = rt27_meas.ant_num obs_trmb['elev'] = obs_gen['elev'] = rt27_meas.elev obs_trmb['azi'] = obs_gen['azi'] = rt27_meas.azi obs_trmb['svFlags'] = [] obs_gen['svFlags'] = [] for flag in rt27_meas.sv_flags: obs_trmb['svFlags'].append(flag) obs_gen['svFlags'].append(flag) obs_trmb['freq'] = obs_gen['freq'] = rt27_meas.freq obs_trmb['track'] = obs_gen['track'] = rt27_meas.track obs_trmb['CNo'] = obs_gen['CNo'] = rt27_meas.cno obs_trmb['pseudo'] = obs_gen['pseudo'] = rt27_meas.pseudo obs_trmb['phase'] = obs_gen['phase'] = rt27_meas.phase obs_trmb['slipCount'] = rt27_meas.slip_count obs_trmb['measFlags'] = [] obs_gen['measFlags'] = [] for flag in rt27_meas.meas_flags: obs_trmb['measFlags'].append(flag) obs_gen['measFlags'].append(flag) obs_trmb['doppler'] = obs_gen['doppler'] = rt27_meas.doppler obs_gen['pseudoValid'] = ((rt27_meas.meas_flags[0] & 0x22) == 0x22) obs_gen['phaseValid'] = ((rt27_meas.meas_flags[0] & 0x01) > 0) obs_gen['dopplerValid'] = ((rt27_meas.meas_flags[0] & 0x04) > 0) obs_gen['cycleSlip'] = ((rt27_meas.meas_flags[0] & 0x10) > 0) return (obs_trmb, obs_gen) class ReturnType(): """ Legacy compatability Define a class to give easy access to the return bitfield for the records decoded this run Note that more that one type may be present """ def __init__(self): self.dNoData = 0 self.dObs = 1 << 0 self.dPVT = 1 << 1 ########################################################################## if __name__ == "__main__": num_args = len(sys.argv) if num_args not in (2, 3, 4): print("Usage:") print(" python RT27.py ") print(" python RT27.py ") print(" python RT27.py ") print(" The system IDs is an optional comma seperated. Only satellite") print(" systems on this list are shown on stdout. e.g.") print(" python RT27.py 10.1.150.xxx 5017 0,2") print(" will output data from GPS and GLONASS. If there are no system IDs,") print(" data from all satellites is output to stdout") sys.exit(1) fptr = None sock = None # https://wiki.eng.trimble.com/index.php/SV_System # Currently there are 0-10 valid systems numSys = 11 showSys = [True] * numSys showFft = True if num_args == 2: # Read binary data from a file fptr = open(sys.argv[1], 'rb') elif( (num_args == 3) or (num_args == 4) ): # Read binary data from a socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.settimeout(10) sock.connect((sys.argv[1], int(sys.argv[2]))) if(num_args == 4): showSys = [False] * numSys tokens = sys.argv[3].split(',') for thisToken in tokens: thisToken = int(thisToken) if(thisToken >= numSys): print("Unknown system:",thisToken) sys.exit(1) showSys[thisToken] = True rt27_parser = ParseRT27() #show_info=True) data_in = [] while fptr is not None or sock is not None: if fptr is not None: MIN_BYTES = 5 * 1024 data_in = bytearray(fptr.read(MIN_BYTES)) if len(data_in) == 0: # End of file, close the file and mark the file as invalid to # terminate the decode loop next time around, giving # RT27.processData() a chance to look at the last data fptr.close() fptr = None elif sock is not None: # Read the next data from the socket data_in = bytearray(sock.recv(5000)) rt27_parser.process_data(data_in) decoded_obs = rt27_parser.get_decoded_obs() for obs in decoded_obs: for meas in obs.meas: sec = obs.hdr.sec sv_id = meas.sv_id sat_type = meas.sat_type elev = meas.elev freq = meas.freq track = meas.track cno = meas.cno slip_count = meas.slip_count mflg0 = meas.meas_flags[0] mflg1 = meas.meas_flags[1] mflg2 = meas.meas_flags[2] mflg3 = meas.meas_flags[3] pseudo = meas.pseudo phase = meas.phase doppler = meas.doppler ant_num = meas.ant_num if sv_id is None: continue if(showSys[sat_type] == True): print(f'{sec:10d} {sv_id:3d} {sat_type:2d} {elev:3d} ', end='') print(f'{freq:2d} {track:2d} {cno:4.1f} {slip_count:3d} ', end='') print(f'{mflg3:02x}{mflg2:02x}{mflg1:02x}{mflg0:02x} ', end='') print(f'{pseudo:12.3f} {phase:14.3f} {doppler:9.3f} {ant_num:d}') decoded_fft = rt27_parser.get_decoded_fft() if showFft: for elem in decoded_fft: for curr_fft in elem.fft: print(f'FFT {curr_fft.freq} {curr_fft.miti} {curr_fft.mode} {curr_fft.ant} {elem.msecs} ' + " ".join([f'{x:.1f}' for x in curr_fft.points]) ) if len(elem.meta.fft_base) > 0: for curr in elem.meta.fft_base: print(f'FFTbase {elem.msecs} {curr[0]} {curr[1]}') if len(elem.meta.gain) > 0: for curr in elem.meta.gain: print(f'FFTgain {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') if len(elem.meta.RSSI) > 0: for curr in elem.meta.RSSI: print(f'FFTrssi {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') if len(elem.meta.quality) > 0: for curr in elem.meta.quality: print(f'FFTqual {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') if len(elem.meta.commonRSSI) > 0: for curr in elem.meta.commonRSSI: print(f'FFTcommonRSSI {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') if len(elem.meta.commonRSSI) > 0: for curr in elem.meta.commonRSSI: print(f'FFTcommonRSSI {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') if len(elem.meta.post_quality) > 0: for curr in elem.meta.post_quality: print(f'FFTpostQual {elem.msecs} {curr[0]} {curr[1]} {curr[2]}') # The legacy data format can also be accessed using: # legacy = rt27_parser.get_legacy_output() # legacy.ret [ParseRT27Legacy.ReturnType().dObs etc.] # legacy.week / secs / bias / numSVs # legacy.obsEpoch / obsGenericEpoch [list of dictionary-based data] # # BUT the legacy format only supports a single observable record per # call to process_data(). Observables can be silently lost, particularly # noticable when decoding from files.