import sys sys.path.append('/afs/slac/g/lcls/package/cpsw/python/lib') import getopt import unittest from cpsw_api_py import * from cpsw_axiv_udp_tst_constants import * #class TestFailed(Exception): # self.assertRaises(TestFailed) def create_axiv( s ): axiv = IMMIODev.create( s, 0x1000, ByteOrder.LE ) f = IIntField.create("dnaValue", 64, False, 0, IIntField.RO, 4) axiv.addAtAddress(f, 0x08) f = IIntField.create("fdSerial", 64, False, 0, IIntField.RO, 4) axiv.addAtAddress(f, 0x10) f = IIntField.create("counter", 32, False, 0, IIntField.RW) axiv.addAtAddress(f, 0x24) f = IIntField.create("bldStamp", 8, False, 0 ) axiv.addAtAddress(f, 0x800, VLEN) f = IIntField.create("scratchPad", 32, True, 0) axiv.addAtAddress(f, 0x04) f = IIntField.create("bits", 22, True, 4) axiv.addAtAddress(f, 0x04) return axiv def create_prbs( s ): prbs = IMMIODev.create( s, 0x1000, ByteOrder.LE ) f = IIntField.create("axiEn", 1, False, 0) prbs.addAtAddress(f, 0x00) f = IIntField.create("trig", 1, False, 1) prbs.addAtAddress(f, 0x00) f = IIntField.create("busy", 1, False, 2, IIntField.RO) prbs.addAtAddress(f, 0x00) f = IIntField.create("overflow", 1, False, 3, IIntField.RO) prbs.addAtAddress(f, 0x00) f = IIntField.create("oneShot", 1, False, 4) prbs.addAtAddress(f, 0x00) f = IIntField.create("packetLength", 32, False) prbs.addAtAddress(f, 0x04) f = IIntField.create("tDest", 8, False, 0) prbs.addAtAddress(f, 0x08) f = IIntField.create("tId", 8, False, 0) prbs.addAtAddress(f, 0x09) f = IIntField.create("dataCnt", 32, False, 4, IIntField.RO) prbs.addAtAddress(f, 0x0C) f = IIntField.create("eventCnt", 32, False, 0, IIntField.RO) prbs.addAtAddress(f, 0x10) f = IIntField.create("randomData", 32, False, 0, IIntField.RO) prbs.addAtAddress(f, 0x14) return prbs def rxThread(): strm = IStream.create( IPath.create("/fpga/dataSource") ) def vpb( l, s ): l.append(s) return s def usage( arg ): print "Usage: ", str(arg), " [-a ] [-mh] [-V ] [-S ] [-n ] [-p ]" print " -a : destination IP" print " -V : SRP version (1 or 2)" print " -m : use 'fake' memory image instead" print " of real device and UDP" print " -S : test streaming interface" print " with frames of 'length'" print " 'length' must be > 0 to enable streaming" print " -n : stream 'shots' fragments" print " (defaults to 10)." print " -p : trigger a fragment every ms" print " (defaults to 1000)." print " -h : print this message" def main(argv): rval = 0 ip_addr = "192.168.2.10" use_mem = False vers = 2 length = 1 shots = 10 period = 1000 #ms try: opts, args = getopt.getopt(argv,"a:mV:S:hn:p:") except getopt.GetoptError: usage( 'h' ) sys.exit(2) for opt, arg in opts: if opt == '-a': ip_addr=str( arg ) elif opt in ("-m"): use_mem = True elif opt in ("-V"): vers = int(arg) elif opt in ("-S"): length = int(arg) elif opt in ("-n"): shots = int(arg) elif opt in ("-p"): period = int(arg) elif opt in ("-h"): usage( arg ) sys.exit(2) else: print 'Unable to scan argument to option ', opt sys.exit(2) if vers != 1 and vers != 2: print 'Invalid protocol version ', vers, ' -- must be 1 or 2' assert ( vers == 1 or vers == 2 ) try: d = IDev.getRootDev() root = INoSsiDev.create("fpga", "192.168.2.19") mmio = IMMIODev.create("mmio", 0x100000) axiv = create_axiv("vers") sysm = IMMIODev.create("sysm", 0x1000, ByteOrder.LE) rmem = IMemDev.create("rmem", 0x100000) prbs = create_prbs("prbs") rmem.addAtAddress( mmio ) sysm.addAtAddress( IIntField.create("adcs", 16, True, 0), 0x400, ADCL, 4) mmio.addAtAddress( axiv, 0x00000 ) mmio.addAtAddress( sysm, 0x10000 ) mmio.addAtAddress( prbs, 0x30000 ) root.addAtAddress( mmio, INoSsiDev.SRP_UDP_V1 if vers == 1 else INoSsiDev.SRP_UDP_V2, 8192, 500 ) d.addAtAddress( root ) pre = IPath.create( root ) bldStamp = IScalVal_RO.create( pre.findByName("mmio/vers/bldStamp") ) fdSerial = IScalVal_RO.create( pre.findByName("mmio/vers/fdSerial") ) dnaValue = IScalVal_RO.create( pre.findByName("mmio/vers/dnaValue") ) counter = IScalVal_RO.create( pre.findByName("mmio/vers/counter" ) ) scratchPad = IScalVal.create ( pre.findByName("mmio/vers/scratchPad") ) bits = IScalVal.create ( pre.findByName("mmio/vers/bits") ) # list for ScalVals vals = [] axiEn = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/axiEn") )) trig = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/trig") )) busy = vpb(vals, IScalVal_RO.create( pre.findByName("mmio/prbs/busy") )) overflow = vpb(vals, IScalVal_RO.create( pre.findByName("mmio/prbs/overflow") )) oneShot = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/oneShot") )) packetLength = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/packetLength") )) tDest = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/tDest") )) tId = vpb(vals, IScalVal.create ( pre.findByName("mmio/prbs/tId") )) dataCnt = vpb(vals, IScalVal_RO.create( pre.findByName("mmio/prbs/dataCnt") )) eventCnt = vpb(vals, IScalVal_RO.create( pre.findByName("mmio/prbs/eventCnt") )) randomData = vpb(vals, IScalVal_RO.create( pre.findByName("mmio/prbs/randomData") )) adcs = IScalVal_RO.create( pre.findByName("mmio/sysm/adcs") ) ## get_value_as * returns tuple with resp[0] is retval from getVal, res[1] is list with values resp = bldStamp.get_value_as_uint8( VLEN ) print 'Build String: ', ''.join( map( chr, resp[1] ) ) resp = fdSerial.get_value_as_uint64() print 'Serial #: ', hex( resp[1][0] ) resp = dnaValue.get_value_as_uint64() print 'DNA: ', hex( resp[1][0] ) resp = counter.get_value_as_uint64() print 'Counter: ', hex( resp[1][0] ) resp = counter.get_value_as_uint64() print 'Counter: ', hex( resp[1][0] ) adcv = adcs.get_value_as_uint16( ADCL ) print '\n\n', 'ADC Values:', '\n' for i in xrange(0, 9): print ' ', hex( (adcv[1])[i] ) u16=0x6765; u32=0xffffffff; scratchPad.set_value_as_uint32( u32 ) bits.set_value_as_uint16( u16 ) resp = scratchPad.get_value_as_uint32() print 'ScratchPad: ', hex( resp[1][0] ) if resp[1][0] == 0xfc06765f: print 'Readback of merged bits (expected 0xfc06765f) PASSED\n' else: print 'Readback of merged bits (expected 0xfc06765f) FAILED\n' #if length > 0: # print 'PRBS Regs:' # for sval in vals: # resp = sval.get_value_as_uint32() # print sval.getName(), ': ', resp[1][0] # v = 1 # axiEn.set_value_as_uint32( v ) # v = length # packetLength.set_value_as_uint32( v ) # v = 1 # oneShot.set_value_as_uint32( v ) # # t = threading.Thread(name='rxThread', target=rxThread) # t.start() # # for i in range(0, shots) # #sleep # oneShot.setVal( 1 ) except: print 'CPSW Error:' if __name__=="__main__": sys.exit(main(sys.argv[1:]))