import socket import sys import os import operator import math import bisect import fnmatch import io from ftplib import FTP from lxml import etree from io import BytesIO import time import requests import struct import zlib import ctypes # note: pycurl doesn't seem to have a timeout for when the server # hangs while sending data. requests does, so use it instead. ###################################################################### # # A collection of useful commands in python for controlling a # Trimble "coreBuild" receiver. There are a combination of Trimcomm, # Programmatic interface and "scrapped" web interface controls. Note # the "scrapped" web interface controls can change as the receiver # web interface evolves. # # Copyright Trimble 2017-2018 ###################################################################### def formDColCommand(dcol_type,dcol_data_list): """Given the type & data of a Trimcomm command, add the header and tail to form a complete packet. Example: # Send Trimcomm 0x51 sub 7 to turn on NMEA GGK on 1st TCP/IP port: cmd=RXTools.formDColCommand(0x51,[7,21,48,3,0]) RXTools.sendDColCommand('10.1.149.xxx',28001,cmd) """ hdr = bytearray([2,0,dcol_type,len(dcol_data_list)]) checksum = dcol_type + len(dcol_data_list) for x in dcol_data_list: checksum += x checksum = checksum & 0xff tail = bytearray([checksum,3]) return hdr + bytearray(dcol_data_list) + tail def getDColSerial( IPAddr, port): cmd=formDColCommand(0x06,[]) ret=sendDColCommand(IPAddr,port,cmd,need_ack=False) print(ret) def sendDColRefWeek( IPAddr, port, oldest_year ): """Send a command to change the reference week # for older RF scenarios. For example, set oldest_year to 2000 to support playback of RF data from Jan 2000. """ import datetime gps_epoch = datetime.datetime(1980,1,6) cmd_epoch = 0 cmd_week = int((datetime.datetime(oldest_year,1,1) - gps_epoch).days/7) while cmd_week >= 1024: cmd_week -= 1024 cmd_epoch += 1 if cmd_week < 0: raise RuntimeError("Impossible year %d" % oldest_year) cmd=formDColCommand(0xA2,[33,cmd_week>>8,cmd_week&0xff,cmd_epoch>>8,cmd_epoch&0xff]) sendDColCommand(IPAddr,port,cmd) # Send a command over DCOL/Trimcomm on a TCP/IP port # Most DCOL commands respond quickly, so have a short 10s timeout. def sendDColCommand(IPAddr,port,command,need_ack=True,timeout=10): data = [] #Initialize a TCP client socket using SOCK_STREAM tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) tcp_client.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if timeout > 0: tcp_client.settimeout(timeout) try: # Establish connection to TCP server tcp_client.connect((IPAddr, port)) tcp_client.sendall(command) data = bytearray(tcp_client.recv(1000)) if(data[0] == 0x06): print("ACK") else: print(list(data)) if need_ack: raise RuntimeError("Didn't get an ACK to command {} IP {} port {}".format(command,IPAddr,port)) except socket.timeout: raise RuntimeError("Didn't get any response to command {} IP {} port {}".format(command,IPAddr,port)) finally: tcp_client.close() return(data) #def sendDColResetCmd(IPAddr,port): # # print IPAddr # print port # # # RTKSTAT Command # #RSTCmd = b"\x02\x00\x32\x02\x01\x01\x36\x03" # # #RTKCTRL Command # RSTCmd = b"\x02\x00\x52\x10\x01\x00\x00\x02\x02\xFF\x00\x00\x01\xBC\x00\x01\x00\x00\x00\x00\x24\x03" # sendDColCommand(IPAddr,port,RSTCmd) class DColSocket: """\ Wrapper to help send/recv DCOL/Trimcomm commands""" def __init__(self, IPAddr, port, timeout=10): """\ IPAddr = receiver address, e.g., '10.1.150.2' port = receiver port, e.g., 5018 """ self.IPAddr = IPAddr self.port = port self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) self.sock.settimeout(timeout) self.sock.connect( (IPAddr,port) ) def send_recv( self, dcol, dcol_data=None): """Returns data from command. If dcol_data is None, then dcol is output from formDColCommand(). Otherwise, dcol == DCOL type and dcol_data = DCOL payload""" if dcol_data is None: cmd = dcol else: cmd = formDColCommand(dcol,dcol_data) self.sock.sendall(cmd) return bytearray(self.sock.recv(256)) def send_recv_ack( self, dcol, dcol_data=None): """Throws error if no ACK. If dcol_data is None, then dcol is output from formDColCommand(). Otherwise, dcol == DCOL type and dcol_data = DCOL payload""" if dcol_data is None: cmd = dcol else: cmd = formDColCommand(dcol,dcol_data) #print("Sending:",['%2.2x'%d for d in cmd]) self.sock.sendall(cmd) data = self.sock.recv(1) if data[0] != 0x06: raise RuntimeError("Didn't get ACK",data) def recv_only(self, nbytes=256): """Grab any data bytes returned to the port.""" return list(self.sock.recv(nbytes)) def close(self,sleep_time=0.5): if self.sock is not None: self.sock.shutdown(socket.SHUT_RDWR) self.sock.close() time.sleep(sleep_time) # make sure socket has time to close self.sock = None def __del__(self): self.close() # Trimcomm 0x53/0xC0 tracking bits in order tracking_defs = [ "GPS_L1P", "GPS_L2P", "GPS_L2", "GPS_L2E_only", "GPS_L2C_ind", "reserved", "GPS_L5", "GLN_G1CA", "GLN_G1P", "GLN_G2P", "GPS_L5_I", "GPS_L5_Q", "GPS_L2CM", "GPS_L2CL", "GLN_G2CA", "GLN_G2CA_fb_G2P", "GAL_E1", "GAL_E5A", "GAL_E5B", "GAL_E5Alt", "GAL_E1_CBOC", "GAL_E1_D", "GAL_E1_P", "GAL_E5A_D", "GAL_E5A_P", "GAL_E5B_D", "GAL_E5B_P", "GAL_E5Alt_D", "GAL_E5Alt_P", "GAL_E5Alt_full", "BDS_B1", "BDS_B2", "BDS_B3", "GPS_L1E", "QZSS_L1CA", "QZSS_L1S", "QZSS_L1C", "QZSS_L1C_D", "QZSS_L1C_P", "QZSS_L2C", "QZSS_L2CM", "QZSS_L2CL", "QZSS_L5", "QZSS_L5_I", "QZSS_L5_Q", "QZSS_LEX", "GLN_G3", "GLN_G3_D", "GLN_G3_P", "GPS_L1CA_off", "IRNSS_L5CA", "IRNSS_S1CA", "GAL_E6", "GAL_E6_D", "GAL_E6_P", "GAL_E6_D_sep_P", "BDS_B1C", "BDS_B1C_MBOC", "BDS_B1C_D", "BDS_B1C_P", "BDS_B2A", "BDS_B2A_D", "BDS_B2A_P", "GPS_L1C", "GPS_L1C_MBOC", "GPS_L1C_D", "GPS_L1C_P", "QZSS_LEX_L6E", "BDS_B2B", ] class TrackingMode: """Fills in tracking_defs[] as attributes into a bit array. Example: x = TrackingMode() print(x) x.set_bit('GPS_L2') print(x.vals) """ def get_n(self,name): for n,val in enumerate(tracking_defs): if val == name: return n raise RuntimeError("Invalid bit: %s"%name) def get_bit(self,name): n = self.get_n(name) return (self.vals[n>>3] >> (n%8))&1 def set_bit(self,name): n = self.get_n(name) self.vals[n>>3] |= 1 << (n%8) def clear_bit(self,name): n = self.get_n(name) self.vals[n>>3] &= (1 << (n%8))^0xff def set_vals(self, vals): if len(vals) < 8: vals += [0]*(8-len(vals)) self.vals = vals def __init__(self, vals=[0]*8): self.set_vals(vals) def __str__(self): val = '' for n,name in enumerate(tracking_defs): if n%8==0: val += "Tracking mode %d:\n"%((n>>3)+1) val += " %d: %s = %d\n"%(n%8,name,self.get_bit(name)) return val class DColTrackConfig: """Use to read and set 0x53 Trimcomm tracking config. Example: x = DColTrackConfig('10.1.2.3',5017) print(x) print(x.mode.vals) x.mode.set_bit('BDS_B1C') # or x.mode.clear_bit('BDS_B1C') x.send_conf() print(x) print(x.mode.vals) The valid bit names are in tracking_defs[] """ def __init__(self, IPAddr, port): self.IPAddr = IPAddr self.port = port self.mode = TrackingMode() self.get_conf() def send_conf(self): try: s = DColSocket( self.IPAddr, self.port ) data = s.send_recv( 0x53, self.mode.vals ) if(data[0] == 0x06): print("OK",data) else: print("Failed to set config",data) finally: del s self.get_conf() def get_conf(self): try: s = DColSocket( self.IPAddr, self.port ) data = s.send_recv( 0xc0, [0xff] ) finally: del s self.mode.set_vals( data[4:-2] ) def __str__(self): return str(self.mode) def getDColStartupData(IPAddr,port): """\ IPAddr = receiver IP address, e.g., '10.1.150.2' port = receiver IP port, e.g., 5018 Return receiver data that can help with a fast startup, e.g., almanac/eph/UTC. Example usage: data = RXTools.getDColStartupData( '10.1.150.2', 5018 ) RXTools.sendDColStartupData( '10.1.150.101', 28001, data ) """ st_data = {'gps_alm':{}, 'gps_eph':{}, 'gps_utc':None, 'gln_alm':{}, 'gln_eph':{}, 'gal_alm':{}, 'gal_eph':{}, 'gal_sys':None, 'bds_alm':{}, 'bds_eph':{}, 'irn_alm':{}, 'irn_eph':{}, 'qzs_alm':{}, 'qzs_eph':{}, 'pos':None } s = DColSocket( IPAddr, port ) def sv_helper( min_sv, max_sv, alm_key, alm_cmd, eph_key, eph_cmd ): for sv in range(min_sv,max_sv+1): data = s.send_recv( 0x54, [alm_cmd, sv, 0] ) if len(data) > 1: st_data[alm_key][sv] = data if eph_key is not None: data = s.send_recv( 0x54, [eph_cmd, sv, 0] ) if len(data) > 1: st_data[eph_key][sv] = data sv_helper( 1, 32, 'gps_alm', 7, 'gps_eph', 1 ) # Get GPS UTC/iono data data = s.send_recv( 0x54, [3,0,0] ) if len(data) > 1: st_data['gps_utc'] = data sv_helper( 1, 24, 'gln_alm', 8, 'gln_eph', 9 ) sv_helper( 1, 36, 'gal_alm', 12, 'gal_eph', 11 ) data = s.send_recv( 0x54, [13, 0, 0] ) if len(data) > 1: st_data['gal_sys'] = data sv_helper( 1, 63, 'bds_alm', 22, 'bds_eph', 21 ) sv_helper( 1, 7, 'irn_alm', 26, 'irn_eph', 25 ) sv_helper( 193, 202, 'qzs_alm', 16, 'qzs_eph', 14 ) # current position data = s.send_recv( 0x30, [0, 0] ) if len(data) > 1: st_data['pos'] = data return st_data def sendDColStartupData(IPAddr, port, st_data, send_time=True): """\ IPAddr = receiver IP address, e.g., '10.1.150.2' port = receiver IP port, e.g., 5018 st_data = result from getDColStartupData() Send receiver data that can help with a fast startup. See getDColStartupData() for more info. Also sends an approximate time based on the current Python UTC time. """ s = DColSocket( IPAddr, port ) def sv_helper( txt, cmd_num, key ): info = [] for sv, data in st_data[key].items(): data = s.send_recv( 0xa9, bytearray([cmd_num, sv]) + data[6:-2] ) if data[0] == 0x6: info.append(sv) print(txt,info) sv_helper("Applied GPS alm for: ", 1, 'gps_alm') sv_helper("Applied GPS eph for: ", 2, 'gps_eph') # Send approximate position if st_data['pos'] is not None: data = s.send_recv( 0xa9, bytearray([5, 0]) + st_data['pos'][9:9+24] ) if data[0] == 0x6: print("last position OK") if send_time: # Send approximate time from datetime import datetime gps_epoch = datetime(1980,1,6) dt = datetime.utcnow() - gps_epoch cmd_week = int(dt.days/7) cmd_secs = int(dt.total_seconds() - cmd_week*60*60*24*7) data = s.send_recv( 0xa9, [0, (cmd_week>>8)&0xff, cmd_week&0xff, (cmd_secs>>24)&0xff, (cmd_secs>>16)&0xff, (cmd_secs>>8)&0xff, cmd_secs&0xff] ) if data[0] == 0x6: print('approx time OK') if st_data['gps_utc'] is not None: # Send GPS UTC/iono data - need to do after approx time # is set to extend the raw UTC week # data = s.send_recv( 0xa9, bytearray([3, 0]) + st_data['gps_utc'][6:-2] ) if data[0] == 0x6: print("GPS UTC/iono OK") # Send eph/alm data for other systems sv_helper("Applied GLN alm for: ", 6, 'gln_alm') sv_helper("Applied GLN eph for: ", 7, 'gln_eph') sv_helper("Applied GAL alm for: ", 0xb, 'gal_alm') sv_helper("Applied GAL eph for: ", 0xc, 'gal_eph') sv_helper("Applied BDS alm for: ", 0xd, 'bds_alm') sv_helper("Applied BDS eph for: ", 0xe, 'bds_eph') sv_helper("Applied IRNSS alm for: ", 0x13, 'irn_alm') sv_helper("Applied IRNSS eph for: ", 0x14, 'irn_eph') sv_helper("Applied QZSS alm for: ", 0x10, 'qzs_alm') sv_helper("Applied QZSS eph for: ", 0x11, 'qzs_eph') def sendDColDisableNV(IPAddr,port,need_ack=True): """Disable Stinger non-volatile updates (e.g., eph/alm flash writes). This setting is not saved across reboots or power cycles.""" cmd = formDColCommand( 0xA2, [34, 1] ) sendDColCommand(IPAddr,port,cmd,need_ack=need_ack) time.sleep(1) def sendAntennaType(IPAddr,port,two_char_ID,serial_num=' '): """Set receiver antenna type. If an antenna type is only 1 character like "Unknown External"=="E", then pass in "E ". """ if len(serial_num) != 8: raise RuntimeError("Invalid serial number length",serial_num) if len(two_char_ID) != 2: raise RuntimeError("Invalid antenna ID length",two_char_ID) two_char_ID = [ord(x) for x in two_char_ID] serial_num = [ord(x) for x in serial_num] cmd = formDColCommand( 0x1b, two_char_ID+serial_num ) sendDColCommand(IPAddr,port,cmd) def sendDColGotoMonitor(IPAddr,port): cmd = formDColCommand( 0x87, [] ) sendDColCommand(IPAddr,port,cmd) def formDColAntennaOffCmd(start_chan=0,end_chan=0x74): return formDColCommand( 0xA2, [5, start_chan, 2, end_chan] ) # effectively switch the antenna off by changing the sample MUX def sendDColAntennaOffCmd(IPAddr,port): print(IPAddr,port,"DColAntennaOff") # Antenna off AntOff = formDColAntennaOffCmd() sendDColCommand(IPAddr,port,AntOff) def formDColAntennaOnCmd(start_chan=0,end_chan=0x74): return formDColCommand( 0xA2, [5, start_chan, 3, end_chan] ) # effectively switch the antenna on by fixing the sample MUX def sendDColAntennaOnCmd(IPAddr,port): print(IPAddr,port,"DColAntennaOn") # Antenna on AntOn = formDColAntennaOnCmd() sendDColCommand(IPAddr,port,AntOn) def sendDColRebootRcvr(IPAddr,port): """Reboot the receiver at the given IP address & TCP/IP port""" print(IPAddr,port,"DColResetRcvr") cmd=formDColCommand(0x58,[0xff, 0x00, ord('R'), ord('E'), ord('S'), ord('E'), ord('T')]) sendDColCommand(IPAddr,port,cmd) def SendHttpGet(IPAddr,loc,user,password, verbose=True ,proxies={}, timeout=10, secure=False): """Get a URL from the receiver through HTTP GET http://IPAddr + loc If proxies={}, use the automatically-detected proxy info (e.g., http_proxy environment variable). If proxies={'http':None}, disable HTTP proxy use. """ if(secure == False): url_str = "http://" + IPAddr + loc else: url_str = "https://" + IPAddr + loc # As the certificate from our receivers can't be authenticated we're going to turn # off validating it. However, this causes a warning so make sure we suppress that. requests.packages.urllib3.disable_warnings() if(verbose): print(url_str) if(secure == False): r = requests.get(url_str, auth=(user,password), proxies=proxies, timeout=timeout) else: # HTTPS request (without validating the certificate) r = requests.get(url_str, auth=(user,password), proxies=proxies, timeout=timeout, verify=False) r.raise_for_status() return r.text # Send a command to the receiver through HTTP POST http://IPAddr + cmdstr def SendHttpPost(IPAddr,cmdstr,user,password, timeout=10): url_str = "http://" + IPAddr + cmdstr r = requests.post(url_str, auth=(user,password), timeout=timeout) r.raise_for_status() return r.text def SendAndCheckHttpPost(IPAddr,cmd,user,password,check_response="1"): response = SendHttpPost(IPAddr,cmd,user,password) if check_response not in response: raise RuntimeError("Web interface returned an error: %s" % response) # Calls SentHttpPost() but retries if there's an exception. The power # strip we are using appears to have a pretty poor TCP/IP stack. Sometimes # an exception is thrown when we attempt to send a command over HTTP. # This function can be used to loop around retrying until whatever issue # the server has is cleared. # Set a default limit # of attempts so we don't hang forever on an error. def SendHttpPostRetry(IPAddr,cmdstr,user,password,timeout=10,n_tries=4): resp = False ret = '' while(resp == False and n_tries > 0): n_tries -= 1 try: ret = SendHttpPost(IPAddr,cmdstr,user,password,timeout=timeout) resp = True except: sys.stderr.write("SendHttpPost - exception trying again\n") time.sleep(1) resp = False return ret # Soft reset def SoftReset(IPAddr,user,password): # Reset the receiver softResetCmd = "/cgi-bin/resetPage.xml?doReset=1" SendHttpPostRetry(IPAddr,softResetCmd,user,password) def clearGNSSReset(IPAddr,user,password): # Clear the eph/almanac and reset softResetCmd = "/cgi-bin/resetPage.xml?doClearGPS=1" SendHttpPostRetry(IPAddr,softResetCmd,user,password) def MSSOnOff(state,IPAddr,user,password): if(state == True): # Enable RTX # serviceType - 2 == RTX # svTRD - 110 == AUTO satellite selection cmd = "/cgi-bin/spot_conf.xml?serviceType=2&preferSource=INTERNAL&extDataMode=0&setting=0&svTRD=110" SendHttpGet(IPAddr,cmd,user,password,verbose=False,timeout=2) else: # Disable MSS cmr = "/cgi-bin/spot_conf.xml?serviceType=0" SendHttpGet(IPAddr,cmd,user,password,verbose=False,timeout=2) def get_ip(): """Get the first routable IP address of the local machine. Returns None if it only finds an address like 127.0.0.1 """ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: # doesn't even have to be reachable s.connect(('10.255.255.255', 1)) IP = s.getsockname()[0] except: IP = None finally: s.close() return IP def ip_to_num(IP_str): """Convert an IP address string like "10.1.2.3" to a number like 0xa010203""" return int(''.join(['%.2X'%x for x in socket.inet_aton(IP_str)]),16) def EnableIpFilter(IPAddr,user,password,bits,target_IP): """Enable IP filtering on a receiver. target_IP = what IP address can access the receiver? 'bits' specifies a mask, so see below for examples. bits = # of mask bits: 8 = mask 255.0.0.0 .. 32 = mask 255.255.255.255 Examples: target_IP = 10.1.2.3 & bits = 32 -> only 10.1.2.3 can access the receiver! target_IP = 10.1.2.3 & bits = 8 -> 10.*.*.* can access the receiver """ if bits < 8 or bits > 32: raise ValueError("Unsupported bits argument",bits) target_IP_num = ip_to_num(target_IP) my_IP_num = ip_to_num( get_ip() ) mask = (0xffffffff << (32 - bits)) & 0xffffffff if (target_IP_num & mask) != (my_IP_num & mask): raise ValueError("target_IP and bits excludes the current machine") cmd = "/xml/dynamic/ipFiltering.xml?enable=1&bits=%d&ip=0x%x" % (bits,target_IP_num) SendHttpGet(IPAddr,cmd,user,password,proxies={'http':None}) def DisableIpFilter(IPAddr,user,password): """Disable receiver IP filtering so that any IP address can connect to receiver""" cmd = "/xml/dynamic/ipFiltering.xml?enable=0" SendHttpGet(IPAddr,cmd,user,password,proxies={'http':None}) def IsTestModeEnabled(IPAddr,user,password): """Check if test mode is enabled on receiver""" txt = SendHttpGet(IPAddr,"/xml/dynamic/sysData.xml",user,password) d = etree.fromstring( txt ) testMode = d.find('testMode') if testMode is not None and testMode.text == 'TRUE': return True return False def getLLH(IPAddr,user,password): """Returns the Lat/Lon/Hgt - None if they can't be extracted""" Week = None Sec = None Lat = None Lon = None Hgt = None try: # First try the programmatic interface as it provides the time # as a floating point number ProgCmd = "/prog/show?position" progData = SendHttpPostRetry(IPAddr,ProgCmd,user,password) for line in progData.splitlines(): if("GpsWeek" in line): Week = int(line.split()[1]) elif("WeekSeconds" in line): Sec = float(line.split()[1]) elif("Latitude" in line): Lat = float(line.split()[1]) elif("Longitude" in line): Lon = float(line.split()[1]) elif("Altitude" in line): Hgt = float(line.split()[1]) if(Week == None): # The programmatic interface wasn't supported (e.g. option bit not on # and not in testmode) # # Grab the position and time from XML. The only problem is we quantize # the time to 1 second txt = SendHttpGet(IPAddr,"/xml/dynamic/posData.xml",user,password,verbose=False) d = etree.fromstring( txt ) Lat = float(d.find('.//lat').text) Lon = float(d.find('.//lon').text) Hgt = float(d.find('.//hgt').text) Sec = float(d.find('.//sec').text) Week = int(d.find('.//week').text) except: pass if(Lat is not None): return(Week,Sec,Lat,Lon,Hgt) else: return(None,None,None,None,None) def DisableTestMode(IPAddr,user,password): """Disable test mode on receiver.""" # Sending any password (other than the testmode PW) will turn off # testmode SendHttpGet(IPAddr,"/cgi-bin/testMode.xml?Password=Anything",user,password) def EnableTestMode(IPAddr,user,password): """Enable test mode on receiver. Some commands below only work in test mode...""" test_password_list = ["TURING","EUCLIDEAN","EUCLID","FARADAY"] for test_pw in test_password_list: if IsTestModeEnabled(IPAddr,user,password): break SendHttpGet(IPAddr,"/cgi-bin/testMode.xml?Password=%s"%test_pw,user,password) # Delete file system # 2018-07-25 this attempts to delete /Internal/ and all # sub-directories. It has been discovered that is dangerous and work # arounds are being investigated. Therefore temporarily removing this # funciton. A better way is to do something like this for a flat # directory structure: # # getFileListAndDelete(IPAddr,user,password,'/Internal/','T02') # getFileListAndDelete(IPAddr,user,password,'/Internal/','T04') # getFileListAndDelete(IPAddr,user,password,'/Internal/FFTs/','png') # getFileListAndDelete(IPAddr,user,password,'/Internal/FFTs/tmp/','png') # # For a session you could do something like the following if you don't # want to parse the directory tree. However, this has an option issue # if the system is logging (the directory can be removed with an open # logging file) # # DeleteDirectory(IPAddr,user,password,'/Internal/DEFAULT') # getFileListAndDelete(IPAddr,user,password,'/Internal/FFTs','png') # getFileListAndDelete(IPAddr,user,password,'/Internal/FFTs/tmp','png') # #def DeleteInternalFiles(IPAddr,user,password): # DeleteFilesCmd = "/xml/dynamic/fileManager.xml?deleteDirectory=/Internal" # SendHttpPostRetry(IPAddr,DeleteFilesCmd,user,password) def DeleteFile(IPAddr,user,password,path,filename): DeleteFileCmd = "/xml/dynamic/fileManager.xml?deleteFiles=" + path + "&f0=" + filename SendHttpPostRetry(IPAddr,DeleteFileCmd,user,password) # Get a directory listing and erase all files with extension "filetype" def getFileListAndDelete(IPAddr,user,password,path,filetype): ret = GetDirectory(IPAddr,user,password,path) for line in ret.splitlines(): if("file" in line): data = line.split() filename = data[1][5:] if( ("." + filetype) in filename): DeleteFile(IPAddr,user,password,path,filename) # Delete Directory def DeleteDirectory(IPAddr,user,password,path): DeleteFilesCmd = "/xml/dynamic/fileManager.xml?deleteDirectory=" + path SendHttpPostRetry(IPAddr,DeleteFilesCmd,user,password) def CreateDummyFile(IPAddr,user,password,path,size,timeout=10): CreateFileCmd = "/prog/set?file&create=yes&checkspace=yes&name=" + path + "&size=" + str(size) SendHttpPostRetry(IPAddr,CreateFileCmd,user,password,timeout) def GetDirectory(IPAddr,user,password,path): getDirCmd = "/prog/show?Directory&path=" + path return SendHttpPostRetry(IPAddr,getDirCmd,user,password) # Delete the GNSS data (Alm/Eph) from the BBFFS def DeleteGNSSData(IPAddr,user,password): DeleteFilesCmd = "/xml/dynamic/fileManager.xml?deleteDirectory=/bbffs/gnssData" SendAndCheckHttpPost(IPAddr,DeleteFilesCmd,user,password,check_response="okay") # see st_iface_defs.h ST_DYN_MOD_* for more info DynamicModelKeyValueText = """ ST_DYN_MOD_NONE = 0, // No dynamic model chosen, so we don't know // the user dynamics explicitly. Stinger will // use a default model that works reasonably // well for low to mid-dynamic applications. ST_DYN_MOD_HUMAN_PORTABLE = 1, // Human walking around ST_DYN_MOD_MAPPING_VEHICLE = 2, // Van or car based mapping system ST_DYN_MOD_OFF_ROAD_VEHICLE = 3, // Off road vehicle ST_DYN_MOD_HEAVY_EQUIPMENT = 4, // Construction equipment // (bulldozer) ST_DYN_MOD_FARM_EQUIPMENT = 5, // Agriculture equipment (tractor) ST_DYN_MOD_AIRBORNE_ROTOR = 6, // Airborne rotor (helicopter) ST_DYN_MOD_AIRBORNE_FIXED = 7, // Airborne fixed wing ST_DYN_MOD_MARINE = 8, // Marine ST_DYN_MOD_RAIL = 9, // Train ST_DYN_MOD_HIGH_DYNAMIC = 10, // High dynamics // (race car, missile?) ST_DYN_MOD_AUTOMOTIVE = 11, // See SPR 10964 to cover mapping vehicle? ST_DYN_MOD_SURVEY_POLE = 12, // Survey pole ST_DYN_MOD_HIGH_VIBRATION = 13, // Construction equipment with high vibration. // Insert new dynamic model types here // Note that the definition of each model should be // sync-ed with factoryInstalledKfDynModels() which is a bit-map // (up to 32 bits). // Per SPR10165, the option bits decides what models are available. // Define the maximum number of suported dynamic models in option bits ST_DYN_MOD_MAX, // A couple of reasons why the Stinger static mode is added here. // The static mode is not defined in the option bits. Since the // option bits are 32-bit long, the Stinger static mode is added // here as 33. Also, the Stinger static mode is needed to match // with the RTK static mode. The downside of this is some special // handling may be needed wherever ST_DYN_MOD_MAX is used to include // ST_DYN_MOD_STATIC. ST_DYN_MOD_STATIC = 33 """ arr = [x.split(',')[0].replace(" ","") for x in DynamicModelKeyValueText.split('\n') if "=" in x ] # eg. arr[i] = "ST_DYN_MOD_MAPPING_VEHICLE=2" arr = ['='.join([y.replace("ST_DYN_MOD_","").replace("_"," ") for y in x.split('=')]) for x in arr] # eg. arr[i] = "MAPPING VEHICLE=2" arr = ['='.join([y.title().replace(" ","") for y in x.split("=")]) for x in arr] # eg. arr[i] = "MappingVehicle=2" mod_name = [x.split("=")[0] for x in arr] mod_num = [int(x.split("=")[1]) for x in arr] DynamicModelKeyValue = dict(zip(mod_num, mod_name)) def GetDynamicModelDict(): return DynamicModelKeyValue # Use web GUI to change dynamic model def ChangeDynamicModel(IPAddr,user,password,modelText): if modelText in DynamicModelKeyValue.values(): for k,v in DynamicModelKeyValue.items(): if modelText == v: model = k else: # see st_iface_defs.h ST_DYN_MOD_* for more info raise RuntimeError("Unknown dynamic model {}".format(modelText)) SendAndCheckHttpPost(IPAddr, '/cgi-bin/positionPage.xml?DynamicModel=%d'%model, user, password) # Use web GUI to get dynamic model def GetDynamicModel(IPAddr,user,password): """return string of model""" raw = SendHttpGet(IPAddr,'/xml/dynamic/configData.xml',user,password) d = etree.fromstring( raw ) DynamicModel_num = int(d.find('.//DynamicModel').text) return DynamicModelKeyValue[DynamicModel_num] # Use web GUI to get elevation mask def GetElev(IPAddr,user,password): """return int of elevMask""" raw = SendHttpGet(IPAddr,'/xml/dynamic/configData.xml',user,password) d = etree.fromstring( raw ) elevMask = int(d.find('.//elevMask').text) return elevMask # Disable logging of the default session def DisableDefaultLogging(IPAddr,user,password): DisableCmd = "/xml/dynamic/dataLogger.xml?disable=DEFAULT" SendAndCheckHttpPost(IPAddr,DisableCmd,user,password) # Sometimes a heavily-loaded receiver can take a little while to # finalize logging. Is there a better way to handle this? time.sleep(5) # Enable logging of the default session def EnableDefaultLogging(IPAddr,user,password): EnableCmd = "/xml/dynamic/dataLogger.xml?enable=DEFAULT" SendAndCheckHttpPost(IPAddr,EnableCmd,user,password) def CheckCloneStatus(IPAddr,user,password): """Clone operations can take a second or two and the status is stored in a global variable. Check the status of any clone file operation. Returns (True, status_xml_string) if everything went OK. Returns (False, status_xml_string) if there was an error. """ count = 0 while True: raw = SendHttpGet(IPAddr,'/xml/dynamic/cloneFileStatus.xml',user,password) d = etree.fromstring( raw ) status = int(d.find('cloneOperationStatus').text) if status == 1: # Clone_Op_InProgWait time.sleep(1) count = count + 1 if count > 20: return (False, raw) continue elif status == 0: # Clone_Op_OK return (True, raw) else: return (False, raw) # Clone the GNSS configuration def CloneGNSSConfig(IPAddr,user,password,filename): filename = filename.strip('.xml') filename = filename.upper() + '.xml' # force to upper case fileroot = filename.strip('.xml') CloneCommand = '/cgi-bin/app_fileUpdate.xml?operation=8&newCloneFileName=' + fileroot + '&cloneAlmEnable=on' SendAndCheckHttpPost(IPAddr,CloneCommand,user,password) result = CheckCloneStatus(IPAddr,user,password) if result[0] == False: raise RuntimeError("Couldn't CloneGNSSConfig: {}".format(result[1])) # Clone all the receiver's configuration def CloneAllConfig(IPAddr,user,password,filename): filename = filename.strip('.xml') filename = filename.upper() + '.xml' # force to upper case fileroot = filename.strip('.xml') CloneCommand = ( '/cgi-bin/app_fileUpdate.xml?operation=8&newCloneFileName=' + fileroot + '&cloneTcpUdpPortEnable=on' + '&cloneEtherBootEnable=on' + '&cloneHttpEnable=on' + '&cloneEmailFtpNtpEnable=on' + '&cloneDataLoggerEnable=on' + '&clonePositionEnable=on' + '&cloneAlmEnable=on' + '&cloneMiscellaneousEnable=on' + '&cloneAllAppfilesEnable=on') SendAndCheckHttpPost(IPAddr,CloneCommand,user,password) result = CheckCloneStatus(IPAddr,user,password) if result[0] == False: raise RuntimeError("Couldn't CloneAllConfig: {}".format(result[1])) # Download the system error log to 'filename' def DownloadSystemErrlog(IPAddr,user,password,filename,timeout=10): # Some firmware versions hang when trying to download an errlog # with nothing in it, so check for that first... url = "http://%s:%s@%s/xml/dynamic/errLog.xml" % (user,password,IPAddr) r = requests.get( url, timeout=timeout ) r.raise_for_status() if 'numEntries>0<' in r.text: open(filename, 'wb').close() return with open(filename,'wb') as f: url = "http://%s:%s@%s/xml/dynamic/SysLog.bin" % (user,password,IPAddr) r = requests.get( url, timeout=timeout ) r.raise_for_status() f.write( r.content ) # Download the cloned GNSS configuration def DownloadClone(IPAddr,user,password,filename,timeout=10): with open(filename,'wb') as f: remote_filename = filename.strip('.xml') remote_filename = remote_filename.upper() + '.xml' # force to upper case remote_filename = remote_filename.split('/')[-1] url = "http://%s:%s@%s/clone_file/%s?gzipFlag=false" % (user,password,IPAddr,remote_filename) r = requests.get( url, timeout=timeout ) r.raise_for_status() f.write( r.content ) with open(filename,'r') as f: line = f.readline() if line.startswith(' 0: try: UploadFile(IPAddr,user,password,url,file_path,response_text="OK:") n_tries = -1 #except RuntimeError(e): except RuntimeError as e: time.sleep(5) n_tries -= 1 if n_tries <= 0: raise RuntimeError(e) # Build firmware for "target" in "directory" def BuildFW(directory,target): command = ('rm -r ' + directory + ' ; cvs co -d ' + directory + ' coreBuild' ' ; cd ' + directory + ' ; ./config ' + target + ' ; make -j8') os.system(command) # Clean up the build directory def CleanFW(directory,target): command = ('cd ' + directory + ' ; ./config ' + target + ' ; make clean') os.system(command) # Install new firmware in a receiver def upgradeFW(IPAddr,user,password,filename,failsafe,wait=False): if(failsafe == True): url = 'http://' + user + ':' + password + '@' + IPAddr + '/prog/Upload?FirmwareFile&failsafe=yes' else: url = 'http://' + user + ':' + password + '@' + IPAddr + '/prog/Upload?FirmwareFile&failsafe=no' print("Upgrade FW: %s to %s" % (filename,url)) UploadFile(IPAddr,user,password,url,filename,response_text="OK:") # In failsafe mode we have the option to wait until done # Don't wait more than 40 minutes - it usually takes ~5-10 minutes. if wait and failsafe: print("Waiting for failsafe to finish...") t_start = time.time() while True: try: raw = SendHttpGet(IPAddr,'/xml/dynamic/firmware_status.xml',user,password,verbose=False) time.sleep(1) d = etree.fromstring( raw ) if int(d.find('in_progress').text) == 0: break if time.time() - t_start > 40*60: raise RuntimeError("Took way too long to upgrade firmware: %s" % raw) except: break # Install the Clone file (that has already been uploaded). # If clear=True, go back to default settings before applying clone # (otherwise you may have random I/O streams enabled after # applying the clone file). def InstallClone(IPAddr,user,password,file_path,clear=False): print("InstallClone: %s" % file_path) path, filename = os.path.split(file_path) upper_file = os.path.splitext(filename)[0].upper() + ".xml" CloneInstall = "/cgi-bin/app_fileUpdate.xml?operation=9&cloneFileName=" + upper_file if clear: CloneInstall += '&clearBeforeInstallCloneFile=1' SendAndCheckHttpPost(IPAddr,CloneInstall,user,password) result = CheckCloneStatus(IPAddr,user,password) if result[0] == False: raise RuntimeError("Couldn't InstallClone: {}".format(result[1])) def modifyCloneAppRec(orig_fname, new_fname, mod_list): """Read orig_fname and write out a new file called 'new_fname' with only the modified data. mod_list = [(app_name, pos_n, pos_len, pos_val), etc.] app_name = some appfile name like "GeneralControls" pos_n = byte offset to modify pos_len = # of bytes to modify pos_val = new value to write For example, mod_list = [('GeneralControls', 38, 1, 2)] will change the dynamic model to 2. Appfiles are generally a stable and consistent receiver interface. """ d = etree.parse(orig_fname) for child in d.getroot(): preserve_elem = False if child.tag == 'SrcReceiverInfo': preserve_elem = True elif child.tag == 'APP_RECORD': for app_name, pos_n, pos_len, pos_val in mod_list: if child.get('name') != app_name: continue data = child.text.replace('\n','').split() big_endian_vals = ['%x' % ((pos_val>>(i*8))&0xff) for i in range(pos_len)] data[pos_n:pos_n+pos_len] = big_endian_vals new_txt = '\n' for i in range(0,len(data),16): new_txt += ' '.join(data[i:i+16]) + ' \n' child.text = new_txt preserve_elem = True if not preserve_elem: child.getparent().remove(child) with io.open(new_fname, 'w', newline='\r\n') as f: f.write(etree.tostring(d,pretty_print=True).decode('utf-8')) def changeCloneDynamicModel( fname, new_fname, dynamicModelStr ): """Change the clone file fname's dynamic model and write only that appfile to new_fname""" dynamicModel = 0 motion = 0 if dynamicModelStr == "Static": dynamicModel = 33 motion = 1 elif dynamicModelStr == "Automotive": dynamicModel = 11 elif dynamicModelStr == "MappingVehicle": dynamicModel = 2 elif dynamicModelStr == "OffRoadVehicle": dynamicModel = 3 elif dynamicModelStr == "Kinematic" or dynamicModelStr == "None": dynamicModel = 0 else: # see st_iface_defs.h ST_DYN_MOD_* for more info raise RuntimeError("Unknown dynamic model {}".format(dynamicModelStr)) modifyCloneAppRec( fname, new_fname, [('GeneralControls', 38, 1, dynamicModel), ('StaticKin', 2, 1, motion)] ) # Disable the NTP Client (used at boot to get time) def DisableNTPClient(IPAddr,user,password): # There is no "disable" for the NTP client, you simply set it up but # without the enable DisableNTP = "/cgi-bin/ntp.xml?ntpServerOffset=0" SendAndCheckHttpPost(IPAddr,DisableNTP,user,password) # Clear the error/warning log def ClearErrorLog(IPAddr,user,password): ClearLog = "/cgi-bin/eraseErrorLog.xml" SendAndCheckHttpPost(IPAddr,ClearLog,user,password) # Get a single file over HTTP def getFileHTTP(IPAddr,user,password,localDir,RXDir,localFile,RXFile): RXDir = RXDir.strip("/") # Strip start / end slashes if they exist with open(localDir+'/'+localFile,"wb") as f: url = "http://%s:%s@%s/download/%s/%s" % (user,password,IPAddr,RXDir,RXFile) r = requests.get( url, timeout=10 ) r.raise_for_status() f.write( r.content ) # 1. Disables the default logging session (the active session is then downloadable) # 2. If localDir doesn't exist it is created # 3. If clean is true deletes all files on the PC in localDir # 4. Downloads all T02 or T04 files from /Internal to localDir # 5. Re-enables logging def GetLoggedFilesViaHttp(IPAddr,user,password,localDir,clean): # First disable logging so that we can download the current file DisableDefaultLogging(IPAddr,user,password) GetLoggedFilesViaHttpNoDisable(IPAddr,user,password,localDir,clean) # re-enable logging EnableDefaultLogging(IPAddr,user,password) # 1. If localDir doesn't exist it is created # 2. If clean is true deletes all files on the PC in localDir # 3. Downloads all T02 or T04 files from /Internal to localDir # # Unlike GetLoggedFilesViaHttp() it does *not* disable the active # logging session. As the active logging file is not a T02 or T04 it # will not attempt to download the open file. # # flat # False - files are all downloaded to this directory localDir + "/" + filename[10:18]. # If the SerialYYYYMMDDHH... format is used this will be localDir/Date/ # True - All files are downloaded to a flat "localDir" directory # # dataDirFormat # False - use the "flat" setting" # True - data is downloaded to filename[10:18] + "/" + localDir. With # the SerialYYYYMMDD .. format maps to Date/localDir - useful # it downloading data from multiple receivers. def GetLoggedFilesViaHttpNoDisable(IPAddr,user,password,localDir,clean,checkLocal=False,flat=True,dateDirFormat=False,timeout=10): if(dateDirFormat == False and not os.path.exists( localDir )): os.makedirs( localDir ) elif(clean == True): # The directory exists - clean it up before we download it for dataFile in os.listdir(localDir): filePath = os.path.join(localDir,dataFile) if((filePath.endswith('.T02') or filePath.endswith('.T04')) and os.path.isfile(filePath)): os.unlink(filePath) raw = SendHttpGet(IPAddr,"/xml/dynamic/fileManager.xml?listDirectory=/Internal",user,password,timeout=timeout,verbose=False) d = etree.fromstring( raw ) fileList = [] for elem in d.iterfind('.//file'): fileList.append([elem.find('.//name').text,int(elem.find('.//size').text)]) # fileList is in a random order, if we have a lot of files # it may take a long while to download the data. Older # files may be deleted before the script attempts to # download them. The minimize the lost data perform a # sort and download the oldest first (assumes files # have date/time encoded in the filename) fileList = sorted(fileList) for filename,size in fileList: if( filename.endswith("T02") or filename.endswith("T04") ): # Only T02 / T04 files if(dateDirFormat): thisDir = filename[10:18] + "/" + localDir elif(flat == False): # Assumes SN+Date format filename! thisDir = localDir + "/" + filename[10:18] else: thisDir = localDir # We'll download the file unless we find out otherwise DownloadFile = True if(checkLocal): # Check if we have the file locally. If we do and the sizes match # don't download it again localFile = thisDir+ '/' + filename if os.path.exists( localFile ): localSize = os.path.getsize( localFile ) else: localSize = -1 print("%s local size %d remote %d"%(filename,localSize,int(size))) if(localSize == size): DownloadFile = False if(DownloadFile == True): string = ("Downloading %s/%s" % (thisDir,filename)) if not os.path.exists( thisDir ): os.makedirs( thisDir ) try: startTime = time.time() with open(thisDir+'/'+filename,"wb") as f: url = "http://%s:%s@%s/download/Internal/%s" % (user,password,IPAddr,filename) r = requests.get( url, timeout=timeout ) r.raise_for_status() f.write( r.content ) stopTime = time.time() # Now get the size from the file system localSize = os.path.getsize( thisDir + '/' + filename) delta = stopTime - startTime if(delta < 0.00001): # Prevent divide by zero (add a little tolerance) delta = 0.00001 rate = (localSize/delta)/(1024*1024) print("%s [size=%d time=%.3f secs, Rate=%.3f MB/sec]" % (string, localSize, delta, rate)) except: print("Problem downloading %s" % filename) # 1. Disables the default logging session (the active session is then downloadable) # 2. If localDir doesn't exist it is created # 3. If clean is true deletes all files on the PC in localDir # 4. Downloads all T02 or T04 files from /Internal to localDir # 5. Re-enables logging def GetLoggedFiles(IPAddr,user,password,localDir,clean): # First disable logging so that we can download the current file DisableDefaultLogging(IPAddr,user,password) ftp = FTP(IPAddr, timeout = 5) ftp.login(user,password) if not os.path.exists( localDir ): os.makedirs( localDir ) elif(clean == True): # The directory exists - clean it up before we download it for dataFile in os.listdir(localDir): filePath = os.path.join(localDir,dataFile) if((filePath.endswith('.T02') or filePath.endswith('.T04')) and os.path.isfile(filePath)): os.unlink(filePath) ftp.sendcmd('CWD /Internal') # Loop around for all files on the server in /Internal. Download if # they are T02 or T04 for remotefile in ftp.nlst(): if(remotefile.endswith('.T02') or remotefile.endswith('.T04')): # Only the T02 or T04's tokens = remotefile.split() fileStr = tokens[-1] print(fileStr) ftpfile = open(localDir + '/' + fileStr, 'wb') status = ftp.retrbinary('RETR ' + fileStr, ftpfile.write) # retrieve file if str(status).startswith('226'): # comes from ftp status: '226 Transfer complete.' print('OK') # print 'OK' if transfer was successful else: print(status) # if error, print retrbinary's return ftpfile.close() ftp.close() # re-enable logging EnableDefaultLogging(IPAddr,user,password) # Similar to GetLoggedFiles, the major difference is this function # compares the files in the local directory and the ones on the receiver # and keeps the localDirectory in sync. It will however not delete any # files on either the receiver or from the local directory. # # If localDir doesn't exist it is created then downloads all # T02 or T04 files from /Internal to localDir if they are # more than 0 bytes long def SyncLoggedFiles(IPAddr,user,password,localDir): ftp = FTP(IPAddr, timeout = 5) ftp.login(user,password) if not os.path.exists( localDir ): os.makedirs( localDir ) h_local_files = [] for file_name in os.listdir(localDir): length = os.path.getsize( localDir + '/' + file_name) fileStr = file_name + " " + str(length) h_local_files.append(fileStr) # populate local dir list ftp.sendcmd('CWD /Internal') h_remote_files = [] for rfile in ftp.nlst(): if(rfile.endswith('.T02') or rfile.endswith('.T04')): # Only the T02 or T04's tokens = rfile.split() fileStr = tokens[-1] + " " + tokens[4]; h_remote_files.append(fileStr) # Remove the local PC files from the list of Receiver files. The # result is a list of files that are on the Receiver but not the PC. # The file name here is the name of the file, followed by a space # and then the size. The name and size have to match otherwise the # script will download the file. This is to handle a situation where # only a partial file gets downloaded. h_diff = sorted(list(set(h_remote_files) - set(h_local_files))) for h in h_diff: tokens = h.split() size = ftp.size(tokens[0]) # Don't bother downloading 0 length files if(size > 0): with open(os.path.join(localDir,tokens[0]), 'wb') as ftpfile: print('Downloading', tokens[0]) try: s = ftp.retrbinary('RETR ' + tokens[0], ftpfile.write) # retrieve file if str(s).startswith('226'): # comes from ftp status: '226 Transfer complete.' print('OK') # print 'OK' if transfer was successful else: print(s) # if error, print retrbinary's return except: print("FTP exception") ftp.close() # Simple function to send one ping to the host and return True if it # gets a response, a time out of 1 second is set. Useful to detect # whether a receiver is on the network def ping(host): """ Returns True if host (str) responds to a ping request. Remember that a host may not respond to a ping (ICMP) request even if the host name is valid. """ # Building the command. Ex: "ping -c 1 google.com" # -W 1 sets the timeout to 1 second command = 'ping -c 1 -W 1 ' + host + " > /dev/null 2>&1" # Pinging return os.system(command) == 0 # Generates a GNSS clone file on receiver at "source", # downloads it to the local PC and then uploads to the # receiver at "dest". Useful to quickly clone the GNSS # almanac/ephemeris data after a receiver has been # cleared. def CloneInstallGNSSConfig(source,sourceUser,sourcePassword, dest,destUser,destPassword): # Generate a clone file on receiver "source" CloneGNSSConfig(source,sourceUser,sourcePassword,"GNSSCLONE.xml") # Download clone file from receiver "source" DownloadClone(source,sourceUser,sourcePassword,"GNSSCLONE.xml") # Upload clone file to receiver "dest" UploadClone(dest,destUser,destPassword,"GNSSCLONE.xml") # Install clone file into receiver "dest" InstallClone(dest,destUser,destPassword,"GNSSCLONE.xml") def deleteNVEph(RXIP,user,password): DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","glnEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","galEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","bdsEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","irnssEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","qzssEph") DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasEph") # In older FW (pre 5.40) the BeiDou file had a different name DeleteFile(RXIP,user,password,"/bbffs/gnssData","compEph") def deleteNVLastPos(RXIP,user,password): DeleteFile(RXIP,user,password,"/bbffs/gnssData","position") def deleteNVAlm(RXIP,user,password,legacyFW): if(legacyFW == False): # This will work for V5.40 and later DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","glnAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","galAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","bdsAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","irnssAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","qzssAlm") DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasAlm") else: # Older FW for sv in range(1,37): DeleteFile(RXIP,user,password,"/bbffs/gnssData","galAlm" + str(sv).zfill(2)) DeleteFile(RXIP,user,password,"/bbffs/gnssData","galAlmHlth" + str(sv).zfill(2)) for sv in range(1,33): DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsAlm" + str(sv).zfill(2)) DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsAlmHealth") DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsAlmConf") for sv in range(1,25): DeleteFile(RXIP,user,password,"/bbffs/gnssData","glnAlm" + str(sv).zfill(2)) for sv in range(1,31): DeleteFile(RXIP,user,password,"/bbffs/gnssData","compAlm" + str(sv).zfill(2)) for sv in range(193,203): DeleteFile(RXIP,user,password,"/bbffs/gnssData","qzssAlm" + str(sv)) for sv in range(120,159): DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasAlm" + str(sv)) DeleteFile(RXIP,user,password,"/bbffs/gnssData","compAlmHealth") DeleteFile(RXIP,user,password,"/bbffs/gnssData","compAlms") DeleteFile(RXIP,user,password,"/bbffs/gnssData","qzssAlmHealth") DeleteFile(RXIP,user,password,"/bbffs/gnssData","glnAlms") DeleteFile(RXIP,user,password,"/bbffs/gnssData","galAlms") def deleteNVUTC(RXIP,user,password): DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","glnUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","galUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","bdsUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","irnssUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","qzssUtc") DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasUtc") def deleteNVIono(RXIP,user,password): DeleteFile(RXIP,user,password,"/bbffs/gnssData","gpsIono") DeleteFile(RXIP,user,password,"/bbffs/gnssData","bdsIonoGrid") DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasIonoBand") # FW older than 5.40 for sv in range(10): DeleteFile(RXIP,user,password,"/bbffs/gnssData","sbasIonoBand" + str(sv)) DeleteFile(RXIP,user,password,"/bbffs/gnssData","compKlobIon") def nv_hash(checksum, x): x = ctypes.c_byte(int(x)).value return (33*checksum + x) & 0xFFFFFFFF # # In /bbffs/ there's a file "position" this # holds the last position sent to stinger. It is # a zipped file once unzipped the buffer is as follows # # CHAR desc[16] # U32 len # U32 checksum # U16 Stinger API Version # U16 Record Version # U32 lla[] num DBLs (sanity check record) # DBL lla[3] ; # DBL drift ; # U8 is_lla_valid ; # U8 is_drift_valid ; # # Given an existing position file this code will # decompress it, modify the lat/lon/hgt and save # a new zipped file. lat/lon should be in degrees, # hgt in meters # def modifyNVPosition(inFile,outFile,lat,lon,hgt): # Decode the input file buf = open(inFile,'rb').read() out = zlib.decompress(bytes(buf)) is_drift_valid = int(out[-1]) is_lla_valid = int(out[-2]) length = struct.unpack('>I',out[16:20])[0] checksum = struct.unpack('>I',out[20:24])[0] stingerAPIVer = struct.unpack('>H',out[24:26])[0] recordVer = struct.unpack('>H',out[26:28])[0] ArrayLen = struct.unpack('>I',out[28:32])[0] drift = struct.unpack('>d',out[-10:-2])[0] hgtIn = struct.unpack('>d',out[-18:-10])[0] lonIn = struct.unpack('>d',out[-26:-18])[0] latIn = struct.unpack('>d',out[-34:-26])[0] print("%f %f %f %f %d %d" % (latIn * 180.0 / math.pi, lonIn * 180.0 / math.pi, hgtIn, drift, is_lla_valid, is_drift_valid)) # Init the checksum calculation checksumComputed = 5381 for i in range(0,16): checksumComputed = nv_hash(checksumComputed,out[i]) # Skop the reported checksum and length for i in range(24,len(out)): checksumComputed = nv_hash(checksumComputed,out[i]) #print("Computed = %x Embedded = %x" % (checksumComputed,checksum)) # Now update out = bytearray(out) # Update the longitude out[-34:-26] = struct.pack('>d',lat*math.pi/180.0) # Update the latitude out[-26:-18] = struct.pack('>d',lon*math.pi/180.0) # Init the checksum calculation checksumComputed = 5381 for i in range(0,16): checksumComputed = nv_hash(checksumComputed,out[i]) # Skop the reported checksum and length for i in range(24,len(out)): checksumComputed = nv_hash(checksumComputed,out[i]) # Compute the updated checksum and update the data buffer out[20:24] = struct.pack('>I',checksumComputed) # Now compress compr = zlib.compress(out) fid = open(outFile,'wb') fid.write(compr) fid.close()