import sys, os import time import logging from collections import namedtuple try: from AgLib.testautomation.drivers.tsip import tsip_core except ImportError: raise ImportError('import tsip_core failed') def translate(s, reverse=False): arr = s.split('\n') rst = {} for l in arr: tp = l.split('\t') if len(tp)<=1: continue k = int(tp[0].strip()) v = '_'.join(tp[1:]) if reverse == False: rst[k] = v else: rst[v] = k return rst def replace_space_with_underscore_in_key(kv): rst = {} for k,v in kv.items(): rst['_'.join(k.split(' '))] = v return rst class AgDevice: ''' Connecting to Ag receiver using TSIP lib via serial Class Tsip() in tsip_core.py is singleton which allows only one serial connection ''' def __init__(self, port): self.tsip = tsip_core.Tsip() self.port = port logging.basicConfig(stream=sys.stdout, level=logging.INFO) # Enable print out of pyflash progress def open_connection(self): print("open port ...") if self.tsip.open_port(self.port): print("open port",self.port,'succeed.') return True else: print("open port",self.port,'failed.') return False def get_sn(self): return self.tsip._get_sn() def get_product_name(self): return self.tsip.get_product_name() def get_product_fw_version(self): return self.tsip.get_fw_version() def get_fw_date(self): incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e1freq, self.tsip.ESP.e45rsp, None) # 0x1f; 0x45 return str("%02d/%02d/%d" % (incomingData.NavProcMonth, incomingData.NavProcDay, incomingData.NavProcYear+1900)) def print_device_info(self): print("----------------------------------------------------------------") print("Using TSIP lib instead of packet cmd ...") print("") product_name = self.get_product_name() print(" device product_name = " + str(product_name) ) product_fw_version = self.get_product_fw_version() print(" device product_fw_version = " + str(product_fw_version) ) product_sn = self.get_sn() print(" device product_sn = " + str(product_sn) ) fw_date = self.get_fw_date() print(" Firmware Date =", fw_date) print("----------------------------------------------------------------") def get_ip_address(self): '''print IP Address''' # cmd only defined in version > 12.0.0.0 # find IP of receiver 0x8e 0xa9 0x07 0x05 try: incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90705req, self.tsip.ESP.e8fa90705rsp, 0x00) if incomingData == None: return "None" else: import ipaddress return '.'.join([x for x in str(ipaddress.ip_address(incomingData.IP)).split('.')[::-1]]) except: return "Failed" def print_ip_address(self): '''print IP Address''' # cmd only defined in version > 12.0.0.0 # find IP of receiver 0x8e 0xa9 0x07 0x05 try: incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90705req, self.tsip.ESP.e8fa90705rsp, 0x00) if incomingData == None: print(" receiver IP request incomingData = ", incomingData) else: import ipaddress print(" receiver IP ", '.'.join([x for x in str(ipaddress.ip_address(incomingData.IP)).split('.')[::-1]]) ) print(" receiver NetMask ", '.'.join([x for x in str(ipaddress.ip_address(incomingData.NetMask)).split('.')[::-1]]) ) print(" receiver Gateway ", '.'.join([x for x in str(ipaddress.ip_address(incomingData.Gateway)).split('.')[::-1]]) ) print("----------------------------------------------------------------") except: print("Error. Unsupported packet cmd: find IP of receiver 0x8e 0xa9 0x07 0x05") def get_tnfs(self): ''' Read TNFS ''' # cmd only defined in version > 12. # Picus TNFS Host Address Info Req try: incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90108req, self.tsip.ESP.e8fa90108rsp, None) return incomingData.IP.decode("utf-8") except: return "Error. Unsupported packet cmd" def print_tnfs(self): ''' Print TNFS ''' print(" tnfs:\t"+self.get_tnfs()) def get_logging(self): ''' Read logging ''' # Picus Logging Control Status Request try: incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90101req, self.tsip.ESP.e8fa90101rsp, None) return str("logging status(0: no logging, 1:logging): %d, filenameLen: %d, full path of the log file: %s ." % ( incomingData.Status, incomingData.filenameLen, str(incomingData.filename.decode("utf-8")).replace('\x00','') ) ) except: return "Error. Unsupported packet cmd" def print_logging(self): ''' Print logging ''' print('print_logging()',self.get_logging()) def get_port_protocol(self,port_n): ''' Print port in and out configuration ''' # Port Configuration port_in_kv = """0 No Input Protocol 1 TAIP Input Protocol 2 TSIP Input Protocol 4 NMEA Input Protocol 8 RTCM Input Protocol 16 DCOL (RTK Link) Input Protocol 32 LBAR Input Protocol 64 CMR Input Protocol 65 NCS Input Protocol 80 Omnistar Demodulated Data Input Protocol 82 AutomaticRTK Input Protocol - RTCM/CMR(+)/CMRx/CMRxe""" port_in_kv = translate(port_in_kv) port_out_kv = """0 No Output Protocol 1 TAIP Output Protocol 2 TSIP Output Protocol 4 NMEA Output Protocol 8 RTCM Output Protocol 8 RTCM Bent Pipe Output Protocol 16 DCOL(RTK Link) Output Protocol 32 LBAR Output Protocol 64 CMR Output Protocol 70 Output ASCII from Port A Protocol 71 Output ASCII from Port B Protocol 81 Omnistar HP Debug Output Protocol""" port_out_kv = translate(port_out_kv) incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.ebcreq, self.tsip.ESP.ebcrsp, port_n) rst = str("Port(%d): Port-%d; InputProtocol-%s; " % (port_n, incomingData.Port, port_in_kv[incomingData.InputProtocol])) rst += str("OutputProtocol-%s; " % port_out_kv[incomingData.OutputProtocol] ) tmp = incomingData.OutputProtocol rst += str("InputBaudRate-%d, OutputBaudRate-%d, DataBits-%d, Parity-%d, StopBits-%d, FlowControl-%d, ProtocolOperation-%d" % (incomingData.InputBaudRate, incomingData.OutputBaudRate, incomingData.DataBits, incomingData.Parity, incomingData.StopBits, incomingData.FlowControl, incomingData.ProtocolOperation)) incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e7a08req, self.tsip.ESP.e7b08rsp, port_n) if tmp == 4: rst += str("; **NMEA: MessageProtocol-%d, MessageInterval-%d, OutputMask-%s, GGAOptions-%d, Precision-%d" % (incomingData.MessageProtocol, incomingData.MessageInterval, "{0:b}".format(incomingData.OutputMask), incomingData.GGAOptions, incomingData.Precision)) return rst def get_all_port_protocol(self): ''' Return all port in and out configuration ''' rst = [] for port_n in [0,1,2]: # 255 is current port rst.append(self.get_port_protocol(port_n)) return rst def print_all_port_protocol(self): ''' Print all port in and out configuration ''' for port_n in [0,1,2]: # print("----------------------------------------------------------------") print( "\n ".join(self.get_port_protocol(port_n).split(';')) ) # print("----------------------------------------------------------------") def soft_reset(self): ''' Send soft reset cmd as reboot ''' incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e25cmd, None, None) print(" Soft Reset. incomingData =", incomingData) return incomingData def set_NEMA_Configuration(self, port_n, msg_interval='Report_at_Position_Rate'): ''' NMEA Extended Port Configuration Options Command ''' message_interval="""0 Report at Position Rate 1 5 seconds 2 1 second 3 200 milliseconds 4 100 milliseconds 255 Retain existing message interval """ print(" config output NEMA at port ", port_n) message_interval = translate(message_interval, True) message_interval = replace_space_with_underscore_in_key(message_interval) incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e7a08cmd, None, port_n, 0, message_interval[msg_interval], 0b1000000, 10, 6) print(" config output NEMA return ", incomingData) return incomingData def set_port(self, port_n, input='TSIP_Input_Protocol', output='TSIP_Output_Protocol', input_baud_rate='115200 bps', output_baud_rate='115200 bps', ): ''' set port in/out protocal, baud rate etc ''' print(" config port ... ", port_n) # Packet Command 0xbc # Serial Port Configuration Parameters Command baud_rate = """0 Don't configure baud rate 1 110 bps 2 300 bps 3 600 bps 4 1200 bps 5 2400 bps 6 4800 bps 7 9600 bps 8 19200 bps 9 38400 bps 10 57600 bps 11 115200 bps""" baud_rate = translate(baud_rate, True) port_in_kv = """0 No Input Protocol 1 TAIP Input Protocol 2 TSIP Input Protocol 4 NMEA Input Protocol 8 RTCM Input Protocol 16 DCOL (RTK Link) Input Protocol 32 LBAR Input Protocol 64 CMR Input Protocol 65 NCS Input Protocol 80 Omnistar Demodulated Data Input Protocol 82 AutomaticRTK Input Protocol - RTCM/CMR(+)/CMRx/CMRxe""" port_in_kv = translate(port_in_kv, reverse=True) port_in_kv = replace_space_with_underscore_in_key(port_in_kv) port_out_kv = """0 No Output Protocol 1 TAIP Output Protocol 2 TSIP Output Protocol 4 NMEA Output Protocol 8 RTCM Output Protocol 8 RTCM Bent Pipe Output Protocol 16 DCOL(RTK Link) Output Protocol 32 LBAR Output Protocol 64 CMR Output Protocol 70 Output ASCII from Port A Protocol 71 Output ASCII from Port B Protocol 81 Omnistar HP Debug Output Protocol""" port_out_kv = translate(port_out_kv, reverse=True) port_out_kv = replace_space_with_underscore_in_key(port_out_kv) # print(port_in_kv) # print(" input:",input,"output:",output) # print(" port_in_kv[input], port_out_kv[output]", port_in_kv[input], port_out_kv[output]) # print(" baud_rate[input_baud_rate], baud_rate[output_baud_rate]", baud_rate[input_baud_rate], baud_rate[output_baud_rate]) # Serial Port Configuration Parameters Command print(" send packet cmd") incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.ebccmd, None, port_n, baud_rate[input_baud_rate], baud_rate[output_baud_rate], 3, 0, 0, 0, port_in_kv[input], port_out_kv[output], 0) # incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.ebcreq, self.tsip.ESP.ebcrsp, port_n) # print(" Now Port(%d): Port-%d, InputProtocol-%s, OutputProtocol-%s" % (port_n, incomingData.Port, port_in_kv[incomingData.InputProtocol], port_out_kv[incomingData.OutputProtocol] )) # print(" InputBaudRate-%d, OutputBaudRate-%d, DataBits-%d, Parity-%d, StopBits-%d, FlowControl-%d, ProtocolOperation-%d" % (incomingData.InputBaudRate, incomingData.OutputBaudRate, incomingData.DataBits, incomingData.Parity, incomingData.StopBits, incomingData.FlowControl, incomingData.ProtocolOperation)) if port_out_kv[output] == 4: time.sleep(5) print(" send packet cmd for NMEA") self.set_NEMA_Configuration(port_n, msg_interval='Report_at_Position_Rate') return incomingData def reset(self): ''' reset ''' incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e1e4bcmd, None, None) print(" Reset Receiver Command incomingData = ", incomingData) return incomingData def set_DGPS_source_control(self, mode='RTK'): ''' set DGPS source control to RTK ''' # DGPS Source Control Command # eg. DGPSSrcMode-6, BeaconAcqMode-0, BeaconFreq0-0, BeaconFreq1-0, BeaconRTCMTimeout-0, SatelliteFrequency-1556000000, SatelliteBitRate-1200, SatelliteRTCMTimeout-20, WAASTimeout-20, CorrectionOptions-0, SatConfigSource-1, mode_num = 6 if mode == 'RTK': mode_num = 6 elif mode == 'RTX': mode_num = 14 incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8e89cmd, None, mode_num,0,0,0,0,1556000000,1200,20,20,0,1) print(" DGPS Source Control Command incomingData = ", incomingData) return incomingData def get_DGPS_source_control(self): ''' get DGPS source control ''' # DGPS Source Control incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8e89req, self.tsip.ESP.e8f89rsp, None) rst = str("DGPSSrcMode-%d, BeaconAcqMode-%d, BeaconFreq0-%d, BeaconFreq1-%d, BeaconRTCMTimeout-%d, SatelliteFrequency-%d, " % (incomingData.DGPSSrcMode, incomingData.BeaconAcqMode, incomingData.BeaconFreq0, incomingData.BeaconFreq1 , incomingData.BeaconRTCMTimeout, incomingData.SatelliteFrequency)) rst += str("SatelliteBitRate-%d, SatelliteRTCMTimeout-%d, WAASTimeout-%d, CorrectionOptions-%d, SatConfigSource-%d, " % (incomingData.SatelliteBitRate, incomingData.SatelliteRTCMTimeout, incomingData.WAASTimeout, incomingData.CorrectionOptions , incomingData.SatConfigSource )) return rst #---------------------------------------------------------------------------- def e35req_e55rsp(self): """[0x35 req --> 0x55 rsp] default setting --> e35cmd_e55rsp(0x1, 0x1, 0x10, 0x2) """ # I/O Option Flag Request incoming_data = self.tsip.generic_packet_handler(self.tsip.ECP.e35req, self.tsip.ESP.e55rsp, None) if incoming_data: incomingData = incoming_data rst = str("PosFlags-%d, VelocityFlags-%d, TimingFlags-%d; AuxFlags-%d" % (incomingData.PosFlags, incomingData.VelocityFlags, incomingData.TimingFlags, incomingData.AuxFlags)) return rst else: return None #---------------------------------------------------------------------------- def e35cmd_e55rsp(self, pos_flags=0x03, vel_flags=0x01, timing_flags=0x11, aux_flags=0x02): """[0x35 cmd --> 0x55 rsp] """ incoming_data = self.tsip.generic_packet_handler(self.tsip.ECP.e35cmd, self.tsip.ESP.e55rsp, pos_flags, vel_flags, timing_flags, aux_flags) if incoming_data: if incoming_data.TimingFlags == timing_flags and incoming_data.PosFlags == pos_flags: incomingData = incoming_data rst = str("PosFlags-%d, VelocityFlags-%d, TimingFlags-%d; AuxFlags-%d" % (incomingData.PosFlags, incomingData.VelocityFlags, incomingData.TimingFlags, incomingData.AuxFlags)) # print(rst) return True else: # print("e35cmd_e55rsp"," was bad") return False else: # print("e35cmd_e55rsp"," failed") return False #---------------------------------------------------------------------------- def get_time(self): ''' get time from receiver ''' # [0x21 req --> 0x41 rsp] rst = "" # test_name = "GPS Time Report Request" incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e21req, self.tsip.ESP.e41rsp, None) if incomingData: rst += str("Time-%f, Week-%d, Offset-%f; " % (incomingData.Time, incomingData.Week, incomingData.Offset)) return rst def get_gps_position_info(self): ''' get GPS Position Packet Pre-request: need to turn on periodic output first using e35cmd_e55rsp() ''' rst = "" # get GPS Position Packet # incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea70100req, self.tsip.ESP.e8fa70100rsp, 0) incomingData = self.tsip.generic_packet_handler(None, self.tsip.ESP.e8fa70100rsp, 0) if incomingData: rst += str("AntennaId-%d, PositionEngine-%d, PositionType-%d, Flags-%d, Latitude-%f, Longitude-%f, Height-%f, NumSatellites-%d, CorrectionAge-%f," % (incomingData.AntennaId, incomingData.PositionEngine, incomingData.PositionType, incomingData.Flags , incomingData.Latitude, incomingData.Longitude, incomingData.Height, incomingData.NumSatellites , incomingData.CorrectionAge )) return rst def get_dgps_corr_info(self): ''' get Differential Correction Information Report ''' # Differential Correction Information Report incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8e9areq, self.tsip.ESP.e8f9arsp, None) rst = str("DataSource-%d, StationID-%d, Age-%.1f, PartialFlag-%d, DGPSFlags-%s, NCSVernierID-%d, RTKFlags-%s," % (incomingData.DataSource, incomingData.StationID, incomingData.Age, incomingData.PartialFlag , "{0:b}".format(incomingData.DGPSFlags), incomingData.NCSVernierID , "{0:b}".format(incomingData.RTKFlags) )) return rst def set_primary_receiver_configuration(self): ''' set Primary Receiver Configuration Parameters Report ''' # Primary Receiver Configuration Parameters Report incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.ebb00cmd, None, 0, 3, 1, 2, 0.0873, 24, 99, 99, 250, 1, 0, 0, 1, 2) return incomingData def get_primary_receiver_configuration(self): ''' get Primary Receiver Configuration Parameters Report ''' # Primary Receiver Configuration Parameters Report incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.ebb00req, self.tsip.ESP.ebb00rsp, None) rst = str("OperatingDimension-%d, DGPSMode-%d, DynamicsCode-%d, SolutionMode-%d, ElevetionMask-%f, AMUMask-%f, PDOP-%f, " % (incomingData.OperatingDimension, incomingData.DGPSMode, incomingData.DynamicsCode, incomingData.SolutionMode , incomingData.ElevetionMask, incomingData.AMUMask, incomingData.PDOP)) rst += str("PDOPSwitch-f-%f, DGPSAgeLimit-%d, FoliageMode-%d, LowPowerMode-%d, ClockHoldMode-%d, MeasurementRate-%d, PosFixRate-%d " % (incomingData.PDOPSwitch, incomingData.DGPSAgeLimit, incomingData.FoliageMode, incomingData.LowPowerMode , incomingData.ClockHoldMode, incomingData.MeasurementRate, incomingData.PosFixRate)) return rst def set_centerpoint_rtx_output_settings(self, datum_id=248): ''' 0x8e 0xab 0x05 Packet Command 0x8e 0xab 0x05 Centerpoint RTX Output Settings ''' # default value is 254 after reset # datum ID 248 effectively disables performing a transform after an RTX position is made. S8eab05cmdRTXOffset_tup = namedtuple('S8eab05cmdRTXOffset_tup', ' enabled east north up') offset = S8eab05cmdRTXOffset_tup._make([0, 0.0, 0.0, 0.0]) # offset = py_tsip.tsip_packets.S8eab05cmdRTXOffset_tup._make([0, 0.0, 0.0, 0.0]) incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8eab05cmd, None, datum_id, offset) return incomingData def get_centerpoint_rtx_output_settings(self): ''' 0x8e 0xab 0x05 Packet Request 0x8e 0xab 0x05 Centerpoint RTX Output Settings ''' incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8eab05req, self.tsip.ESP.e8fab05rsp, None) rst = str("RTXOutputDatum-%d, " % (incomingData.RTXOutputDatum)) rst += str("RTXOffset-enabled-%d, RTXOffset-east-%f, RTXOffset-north-%f, RTXOffset-up-%f" % (incomingData.RTXOffset.enabled, incomingData.RTXOffset.east, incomingData.RTXOffset.north, incomingData.RTXOffset.up)) return rst def download(self, job_key='AGG-FUS3138-2' ): ''' download shift return shifter filename ''' try: from AgLib.testautomation.utils.firmware_util import bamboo_util except ImportError: raise ImportError('import bamboo_util failed') USER = "svcacct_ag_test" try: BAMBOO_PW = os.environ['BAMBOO_PW'] print(BAMBOO_PW) except Exception as e: print("Unknown BAMBOO_PW. Abort.") raise RuntimeError('Unknown BAMBOO_PW. Abort.') flavor = 'do not care' device_type = 'NAV-900' print("Downloading ...", job_key) bamboo = bamboo_util.Bamboo(USER, BAMBOO_PW, job_key) artifact_links = bamboo.get_artifacts_link(job_key, device_type, flavor) print("url:",artifact_links) build_names = bamboo.download_build(artifact_links) print("build_names:",build_names) print("Downloading Done") return build_names def install(self, build_name="shifter_12_00_500_AGFUSION_4170_hard_disable_rtc_6c0f6c9a7f_047_0.img" ): ''' install shift return whether successfully installed ''' try: print("Installing ...",build_name) install = self.tsip.install_build(build_name) return True except TypeError: print('Failed to install builds. Ending test') return False def set_logging(self, turn_on=True, log_filename="pref"): ''' turn on/off logging''' if turn_on: # Packet Command 0x8e 0xa9 0x01 0x01 # Picus Logging Control Command # 0:start, 1:stop print(" start logging ... ") log_filename_bytes = bytearray(log_filename, 'ascii') incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90101cmd, None, 0x00, len(log_filename), log_filename_bytes) print(" enable logging incomingData = ", incomingData) return incomingData else: print(" stop logging ... ") log_filename_bytes = bytearray(log_filename, 'ascii') incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90101cmd, None, 0x01, len(log_filename), log_filename_bytes) print(" stop logging incomingData = ", incomingData) return incomingData def set_tnfs(self, tnfs_ip_addr = "10.1.148.239"): ''' set TNFS server ip address Receiver will test tnfs server and write log file to the root of server ''' print(" TNFS set ... ",tnfs_ip_addr) tnfs_ip_addr_bytes = bytearray(tnfs_ip_addr, 'ascii') incomingData = self.tsip.generic_packet_handler(self.tsip.ECP.e8ea90108cmd, None, tnfs_ip_addr_bytes) print(" TNFS set incomingData = ", incomingData) return incomingData def close_connection(self): ''' Tsip:Closing Serial Port Return True/False ''' try: self.tsip.close_port() time.sleep(2) return True except: return False if __name__ == "__main__": # eg. >>python3 AGRXTools.py /dev/ttyr03 download if len(sys.argv) == 1: print("eg. python3a test_3c.py /dev/ttyr03 [set_datum] [skip_print] [soft_reset] [set_tnfs/unset_tnfs] [start_logging/stop_logging] [port,number,Input Protocol,Output Protocol eg. port,1,No_Input_Protocol,NMEA_Output_Protocol] [rest] [RTK] [download] [install] [debug]") exit(0) print("Connecting to ", sys.argv[1],"...") dev = AgDevice(sys.argv[1]) dev.open_connection() if 'set_datum' in sys.argv: print("Run set_centerpoint_rtx_output_settings()",dev.set_centerpoint_rtx_output_settings(248)) if "skip_print" not in sys.argv: print("Check logging:", dev.get_logging()) dev.print_device_info() dev.print_tnfs() dev.print_logging() dev.print_ip_address() # arg = "port,2,AutomaticRTK_Input_Protocol_-_RTCM/CMR(+)/CMRx/CMRxe,No_Output_Protocol" # _, port_n, input, output = arg.split(',') # port_n = int(port_n) # dev.set_port(port_n, input, output) dev.print_all_port_protocol() # print("Set DGPS config using RTK(DGPSSrcMode==14)") # dev.set_DGPS_source_control('RTX') print("Check get_centerpoint_rtx_output_settings()",dev.get_centerpoint_rtx_output_settings()) print("Check DGPS config using RTK(DGPSSrcMode==6):",dev.get_DGPS_source_control()) print("Check GPS config Pos Rate(PosFixRate==2):",dev.get_primary_receiver_configuration()) print() if "RTK" in sys.argv: dev.set_DGPS_source_control() if "soft_reset" in sys.argv: dev.soft_reset() for arg in sys.argv: if "port" in arg: # eg. port,2,TSIP_Input_Protocol,NMEA_Output_Protocol _, port_n, input, output = arg.split(',') port_n = int(port_n) dev.set_port(port_n, input, output) print("After setting port",port_n) print(dev.get_port_protocol(port_n)) if "reset" in sys.argv: dev.reset() if "download" in sys.argv: filename = dev.download('AGG-FUS3138-2') if "install" in sys.argv: dev.install(filename) else: if "install" in sys.argv: import logging logging.basicConfig( level=getattr(logging, 'INFO'), format='[%(asctime)-15s %(levelname)-8s] %(message)s') fn = 'Builds/RX32-coreBuild/shifter_12_20_500_AGFUSION_4170_hard_disable_rtc_16ed7f63f_008_0.img' dev.install(fn) if "start_logging" in sys.argv: dev.set_logging(True, "prefix") if "stop_logging" in sys.argv: dev.set_logging(False, "stop") if "set_tnfs" in sys.argv: dev.set_tnfs("10.1.148.239") if "unset_tnfs" in sys.argv: dev.set_tnfs("0.0.0.0") if "rtkinfo" in sys.argv: # print(dev.get_dgps_corr_info()) flag = 3 if flag == 0: # print only print(dev.e35req_e55rsp()) print(dev.get_gps_position_info()) elif flag == 1: # turn on periodic output print(dev.e35req_e55rsp()) print(dev.e35cmd_e55rsp()) print(dev.get_gps_position_info()) elif flag == 2: # turn off periodic output print(dev.e35req_e55rsp()) print(dev.e35cmd_e55rsp(0x1, 0x1, 0x10, 0x2)) print(dev.e35req_e55rsp()) print(dev.get_gps_position_info()) elif flag == 3: # turn on, print, then off periodic output print(dev.e35cmd_e55rsp()) print(dev.get_gps_position_info()) print(dev.e35cmd_e55rsp(0x1, 0x1, 0x10, 0x2)) print(dev.get_dgps_corr_info()) if "debug" in sys.argv: print("init setting as RFReplay ...") import time dt = 7 print("Reset ...") dev.reset() print("sleep 30 sec") time.sleep(30) print("Re-init ...") dev = AgDevice(sys.argv[1]) arg = "port,1,No_Input_Protocol,NMEA_Output_Protocol" _, port_n, input, output = arg.split(',') port_n = int(port_n) dev.set_port(port_n, input, output) time.sleep(dt) arg = "port,2,RTCM_Input_Protocol,No_Output_Protocol" _, port_n, input, output = arg.split(',') port_n = int(port_n) dev.set_port(port_n, input, output) time.sleep(dt) rt = dev.set_DGPS_source_control() print("set_DGPS_source_control() returns",rt) time.sleep(dt) rt = dev.set_primary_receiver_configuration() print("set_primary_receiver_configuration() returns", rt) time.sleep(dt) rt = dev.set_logging(True, "RX00") time.sleep(dt) print("set logging",dev.get_logging()) rt = dev.set_tnfs() time.sleep(dt) print(dev.get_tnfs()) # check # dev.print_all_port_protocol() print("Check DGPS config using RTK(DGPSSrcMode==6):",dev.get_DGPS_source_control()) print("Check GPS config Pos Rate(PosFixRate==2):",dev.get_primary_receiver_configuration()) print() rt = dev.set_logging(False, "RX00") time.sleep(dt) print("unset logging",dev.get_logging()) rt = dev.close_connection() print("Close connection =",rt) print("sleep 30 sec") time.sleep(30)