Commit a216d4a9 authored by Javier Díaz's avatar Javier Díaz

file cleaning

parent f00d7ff1
#!/usr/bin/python
import sys
import rr
import time
class CCSR:
def __init__(self, bus, base_addr):
self.base_addr = base_addr;
self.bus = bus;
def wr_reg(self, addr, val):
#print(" wr:%.8X reg:%.8X")%(val,(self.base_addr+addr))
self.bus.iwrite(0, self.base_addr + addr, 4, val)
def rd_reg(self, addr):
reg = self.bus.iread(0, self.base_addr + addr, 4)
#print(" reg:%.8X value:%.8X")%((self.base_addr+addr), reg)
return reg
def wr_bit(self, addr, bit, value):
reg = self.rd_reg(addr)
if(0==value):
reg &= ~(1<<bit)
else:
reg |= (1<<bit)
self.wr_reg(addr, reg)
def rd_bit(self, addr, bit):
if(self.rd_reg(addr) & (1<<bit)):
return 1
else:
return 0
#!/usr/bin/python
import rr
import struct
import time
import sys
from xwb_gpio import *
from i2c import *
from eeprom_24aa64 import *
from onewire import *
from ds18b20 import *
class VIC_irq:
def __init__(self, bus, base):
self.bus = bus;
self.base = base;
def set_reg(self, adr, value):
self.bus.iwrite(0, self.base + adr, 4, value)
def get_reg(self, adr):
return self.bus.iread(0, self.base + adr, 4)
class wrcore_time:
def __init__(self, bus, base):
self.bus = bus;
self.base = base;
def set_reg(self, adr, value):
self.bus.iwrite(0, self.base + adr, 4, value)
def get_reg(self, adr):
return self.bus.iread(0, self.base + adr, 4)
class CDAC5578:
CMD_POWER_ON = 0x40
CMD_WRITE_CH = 0x30
CMD_SW_RESET = 0x70
CMD_LDAC_CTRL = 0x60
CMD_READ_REG = 0x10
def __init__(self, bus, addr):
self.bus = bus;
self.addr = addr;
self.cmd_out(self.CMD_SW_RESET, 0);
self.cmd_out(self.CMD_LDAC_CTRL, 0xff); # ignore LDAC pins
self.cmd_out(self.CMD_POWER_ON, 0x1f, 0xe0);
def cmd_out(self, cmd, data, data2 = 0):
self.bus.start(self.addr, True);
self.bus.write(cmd, False)
self.bus.write(data, False)
self.bus.write(data2, True)
def cmd_in(self, cmd):
self.bus.start(self.addr, True)
self.bus.write(cmd, False)
self.bus.start(self.addr, False)
reg_val=self.bus.read(False)
self.bus.read(True)
return(reg_val)
def out(self, channel, data):
self.cmd_out(self.CMD_WRITE_CH | channel, data)
def rd_out(self, channel):
return(self.cmd_in(self.CMD_READ_REG | channel))
class CFmcDio:
BASE_ONEWIRE = 0x0
BASE_I2C = 0x100
BASE_GPIO = 0x200
BASE_REGS = 0x300
I2C_ADDR_DAC = 0x48
I2C_ADDR_EEPROM = 0x50
I2C_PRESCALER = 400
PIN_PRSNT = 30
GPIO_LED_TOP = 27
GPIO_LED_BOTTOM = 28
DAC_CHANNEL_CORRESP=[0,1,2,7,4,5]
def __init__(self, bus, base):
self.bus = bus;
self.gpio = CGPIO(bus, base + self.BASE_GPIO)
if(not self.fmc_present()):
raise DeviceNotFound("FMC", 0x60000)
self.i2c = COpenCoresI2C(bus, base + self.BASE_I2C, self.I2C_PRESCALER)
self.onewire = COpenCoresOneWire(self.bus, base + self.BASE_ONEWIRE, 624/2, 124/2)
self.eeprom = C24AA64(self.i2c, self.I2C_ADDR_EEPROM);
self.dac = CDAC5578(self.i2c, self.I2C_ADDR_DAC);
self.ds1820 = CDS18B20(self.onewire, 0);
self.i2c.scan()
def fmc_present(self):
return not self.gpio.inp(self.PIN_PRSNT);
def set_dir(self, port, d):
self.gpio.outp(port * 4 + 1, not d)
def set_out(self, port, d):
self.gpio.outp(port * 4, d)
def set_term(self, port, d):
self.gpio.outp(port * 4 + 2, d)
def get_in(self, port):
return self.gpio.inp(port * 4)
def power(self, ins, clock):
pass
def set_led(self, led, state):
gpio_leds=[self.GPIO_LED_TOP, self.GPIO_LED_BOTTOM]
self.gpio.outp(gpio_leds[led],state)
#print "LED", gpio_leds[led], "set to", state
def get_unique_id(self):
return self.ds1820.read_serial_number()
def set_in_threshold(self, port, threshold):
self.dac.out(self.DAC_CHANNEL_CORRESP[port], threshold)
def get_in_threshold(self, port):
return(self.dac.rd_out(self.DAC_CHANNEL_CORRESP[port]))
def get_temp(self):
serial_number = self.ds1820.read_serial_number()
if(serial_number == -1):
return -1
else:
return self.ds1820.read_temp(serial_number)
def set_reg(self, adr, value):
self.bus.iwrite(0, 0x62000 + self.BASE_REGS + adr, 4, value)
def get_reg(self, adr):
return self.bus.iread(0, 0x62000 + self.BASE_REGS + adr, 4)
def get_reg_long(self, adr):
return self.bus.iread(0, 0x62000 + self.BASE_REGS + adr, 8)
def wait_irq_spec(self):
return self.bus.irqwait()
#spec = rr.Gennum()
#dio= CFmcDio(spec, 0x80000);
#print("S/N: %x" % dio.get_unique_id())
#print("Board temp: %d degC" % dio.get_temp());
#!/usr/bin/python
import sys
import rr
import time
import onewire
class CDS18B20:
# ROM commands
ROM_SEARCH = 0xF0
ROM_READ = 0x33
ROM_MATCH = 0x55
ROM_SKIP = 0xCC
ROM_ALARM_SEARCH = 0xEC
# DS18B20 fonctions commands
CONVERT_TEMP = 0x44
WRITE_SCRATCHPAD = 0x4E
READ_SCRATCHPAD = 0xBE
COPY_SCRATCHPAD = 0x48
RECALL_EEPROM = 0xB8
READ_POWER_SUPPLY = 0xB4
# Thermometer resolution configuration
RES = {'9-bit':0x0, '10-bit':0x1, '11-bit':0x2, '12-bit':0x3}
def __init__(self, onewire, port):
self.onewire = onewire
self.port = port
def read_serial_number(self):
#print('[DS18B20] Reading serial number')
if(1 != self.onewire.reset(self.port)):
print('[DS18B20] No presence pulse detected')
return None
else:
#print('[DS18B20] Write ROM command %.2X') % self.ROM_READ
err = self.onewire.write_byte(self.port, self.ROM_READ)
if(err != 0):
print('[DS18B20] Write error')
return None
family_code = self.onewire.read_byte(self.port)
serial_number = 0
for i in range(6):
serial_number |= self.onewire.read_byte(self.port) << (i*8)
crc = self.onewire.read_byte(self.port)
#print('[DS18B20] Family code : %.2X') % family_code
#print('[DS18B20] Serial number: %.12X') % serial_number
#print('[DS18B20] CRC : %.2X') % crc
return ((crc<<56) | (serial_number<<8) | family_code)
def access(self, serial_number):
#print('[DS18B20] Accessing device')
if(1 != self.onewire.reset(self.port)):
print('[DS18B20] No presence pulse detected')
return -1
else:
#print('[DS18B20] Write ROM command %.2X') % self.ROM_MATCH
err = self.onewire.write_byte(self.port, self.ROM_MATCH)
#print serial_number
block = []
for i in range(8):
block.append(serial_number & 0xFF)
serial_number >>= 8
#print block
self.onewire.write_block(self.port, block)
return 0
def read_temp(self, serial_number):
#print('[DS18B20] Reading temperature')
err = self.access(serial_number)
#print('[DS18B20] Write function command %.2X') % self.CONVERT_TEMP
err = self.onewire.write_byte(self.port, self.CONVERT_TEMP)
time.sleep(1)
err = self.access(serial_number)
#print('[DS18B20] Write function command %.2X') % self.READ_SCRATCHPAD
err = self.onewire.write_byte(self.port, self.READ_SCRATCHPAD)
data = self.onewire.read_block(self.port, 9)
#for i in range(9):
# print('Scratchpad data[%1d]: %.2X') % (i, data[i])
temp = (data[1] << 8) | (data[0])
if(temp & 0x1000):
temp = -0x10000 + temp
temp = temp/16.0
return temp
# Set temperature thresholds
# Configure thermometer resolution
#!/usr/bin/python
import sys
import rr
import time
import i2c
class C24AA64:
def __init__(self, i2c, i2c_addr):
self.i2c = i2c
self.i2c_addr = i2c_addr
def wr_data(self, mem_addr, data):
self.i2c.start(self.i2c_addr, True)
self.i2c.write((mem_addr >> 8), False)
self.i2c.write((mem_addr & 0xFF), False)
#print('24AA64:write: data lenght=%d')%(len(data))
for i in range(len(data)-1):
#print('24AA64:write: i=%d')%(i)
self.i2c.write(data[i],False)
i += 1
#print('24AA64:write:last i=%d')%(i)
self.i2c.write(data[i],True)
return 0;
def rd_data(self, mem_addr, size):
self.i2c.start(self.i2c_addr, True)
self.i2c.write((mem_addr >> 8), False)
self.i2c.write((mem_addr & 0xFF), False)
self.i2c.start(self.i2c_addr, False)
data = []
#print('24AA64:read: data lenght=%d')%(size)
i=0
for i in range(size-1):
data.append(self.i2c.read(False))
#print('24AA64:read: i=%d')%(i)
i += 1
#print('24AA64:read:last i=%d')%(i)
data.append(self.i2c.read(True))
return data;
#!/usr/bin/python
import sys
import rr
import time
import csr
class CGN4124:
# Host registers (BAR12), for DMA items storage
HOST_BAR = 0xC
HOST_DMA_CARRIER_START_ADDR = 0x00
HOST_DMA_HOST_START_ADDR_L = 0x04
HOST_DMA_HOST_START_ADDR_H = 0x08
HOST_DMA_LENGTH = 0x0C
HOST_DMA_NEXT_ITEM_ADDR_L = 0x10
HOST_DMA_NEXT_ITEM_ADDR_H = 0x14
HOST_DMA_ATTRIB = 0x18
# GN4124 chip registers (BAR4)
GN4124_BAR = 0x4
R_CLK_CSR = 0x808
R_INT_CFG0 = 0x820
R_GPIO_DIR_MODE = 0xA04
R_GPIO_INT_MASK_CLR = 0xA18
R_GPIO_INT_MASK_SET = 0xA1C
R_GPIO_INT_STATUS = 0xA20
R_GPIO_INT_VALUE = 0xA28
CLK_CSR_DIVOT_MASK = 0x3F0
INT_CFG0_GPIO = 15
GPIO_INT_SRC = 8
# GN4124 core registers (BAR0)
R_DMA_CTL = 0x00
R_DMA_STA = 0x04
R_DMA_CARRIER_START_ADDR = 0x08
R_DMA_HOST_START_ADDR_L = 0x0C
R_DMA_HOST_START_ADDR_H = 0x10
R_DMA_LENGTH = 0x14
R_DMA_NEXT_ITEM_ADDR_L = 0x18
R_DMA_NEXT_ITEM_ADDR_H = 0x1C
R_DMA_ATTRIB = 0x20
DMA_CTL_START = 0
DMA_CTL_ABORT = 1
DMA_CTL_SWAP = 2
DMA_STA = ['Idle','Done','Busy','Error','Aborted']
DMA_ATTRIB_LAST = 0
DMA_ATTRIB_DIR = 1
def rd_reg(self, bar, addr):
return self.bus.iread(bar, addr, 4)
def wr_reg(self, bar, addr, value):
self.bus.iwrite(bar, addr, 4, value)
def __init__(self, bus, csr_addr):
self.bus = bus
self.dma_csr = csr.CCSR(bus, csr_addr)
self.dma_item_cnt = 0
# Get page list
self.pages = self.bus.getplist()
# Shift by 12 to get the 32-bit physical addresses
self.pages = [addr << 12 for addr in self.pages]
self.set_interrupt_config()
# Enable interrupt from gn4124
self.bus.irqena()
# Set local bus frequency
def set_local_bus_freq(self, freq):
# freq in MHz
# LCLK = (25MHz*(DIVFB+1))/(DIVOT+1)
# DIVFB = 31
# DIVOT = (800/LCLK)-1
divot = int(round((800/freq)-1,0))
#print '%d' % divot
data = 0xe001f00c + (divot << 4)
#print '%.8X' % data
#print 'Set local bus freq to %dMHz' % int(round(800/(divot+1),0))
self.wr_reg(self.GN4124_BAR, self.R_CLK_CSR, data)
# Get local bus frequency
# return: frequency in MHz
def get_local_bus_freq(self):
reg = self.rd_reg(self.GN4124_BAR, self.R_CLK_CSR)
divot = ((reg & self.CLK_CSR_DIVOT_MASK)>>4)
return (800/(divot + 1))
# Get physical addresses of the pages allocated to GN4124
def get_physical_addr(self):
return self.pages
# Wait for interrupt
def wait_irq(self):
# Add here reading of the interrupt source (once the irq core will be present)
return self.bus.irqwait()
# GN4124 interrupt configuration
def set_interrupt_config(self):
# Set interrupt line from FPGA (GPIO8) as input
self.wr_reg(self.GN4124_BAR, self.R_GPIO_DIR_MODE, (1<<self.GPIO_INT_SRC))
# Set interrupt mask for all GPIO except for GPIO8
self.wr_reg(self.GN4124_BAR, self.R_GPIO_INT_MASK_SET, ~(1<<self.GPIO_INT_SRC))
# Make sure the interrupt mask is cleared for GPIO8
self.wr_reg(self.GN4124_BAR, self.R_GPIO_INT_MASK_CLR, (1<<self.GPIO_INT_SRC))
# Interrupt on rising edge of GPIO8
self.wr_reg(self.GN4124_BAR, self.R_GPIO_INT_VALUE, (1<<self.GPIO_INT_SRC))
# GPIO as interrupt 0 source
self.wr_reg(self.GN4124_BAR, self.R_INT_CFG0, (1<<self.INT_CFG0_GPIO))
# Get DMA controller status
def get_dma_status(self):
reg = self.dma_csr.rd_reg(self.R_DMA_STA)
if(reg > len(self.DMA_STA)):
print("DMA status register : %.8X") % reg
raise Exception('Invalid DMA status')
else:
return self.DMA_STA[reg]
# Configure DMA byte swapping
# 0 = A1 B2 C3 D4 (straight)
# 1 = B2 A1 D4 C3 (swap bytes in words)
# 2 = C3 D4 A1 B2 (swap words)
# 3 = D4 C3 B2 A1 (invert bytes)
def set_dma_swap(self, swap):
if(swap > 3):
raise Exception('Invalid swapping configuration : %d') % swap
else:
self.dma_csr.wr_reg(self.R_CTL, (swap << self.DMA_CTL_SWAP))
# Add DMA item (first item is on the board, the following in the host memory)
# carrier_addr, host_addr, length and next_item_addr are in bytes
# dma_dir = 1 -> PCIe to carrier
# dma_dir = 0 -> carrier to PCIe
# dma_last = 0 -> last item in the transfer
# dma_last = 1 -> more item in the transfer
# Only supports 32-bit host address
def add_dma_item(self, carrier_addr, host_addr, length, dma_dir, last_item):
if(0 == self.dma_item_cnt):
# write the first DMA item in the carrier
self.dma_csr.wr_reg(self.R_DMA_CARRIER_START_ADDR, carrier_addr)
self.dma_csr.wr_reg(self.R_DMA_HOST_START_ADDR_L, (host_addr & 0xFFFFFFFF))
self.dma_csr.wr_reg(self.R_DMA_HOST_START_ADDR_H, (host_addr >> 32))
self.dma_csr.wr_reg(self.R_DMA_LENGTH, length)
self.dma_csr.wr_reg(self.R_DMA_NEXT_ITEM_ADDR_L, (self.pages[0] & 0xFFFFFFFF))
self.dma_csr.wr_reg(self.R_DMA_NEXT_ITEM_ADDR_H, 0x0)
attrib = (dma_dir << self.DMA_ATTRIB_DIR) + (last_item << self.DMA_ATTRIB_LAST)
self.dma_csr.wr_reg(self.R_DMA_ATTRIB, attrib)
else:
# write nexy DMA item(s) in host memory
# uses page 0 to store DMA items
# current and next item addresses are automatically set
current_item_addr = (self.dma_item_cnt-1)*0x20
next_item_addr = (self.dma_item_cnt)*0x20
self.wr_reg(self.HOST_BAR, self.HOST_DMA_CARRIER_START_ADDR + current_item_addr, carrier_addr)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_HOST_START_ADDR_L + current_item_addr, host_addr)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_HOST_START_ADDR_H + current_item_addr, 0x0)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_LENGTH + current_item_addr, length)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_NEXT_ITEM_ADDR_L + current_item_addr,
self.pages[0] + next_item_addr)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_NEXT_ITEM_ADDR_H + current_item_addr, 0x0)
attrib = (dma_dir << self.DMA_ATTRIB_DIR) + (last_item << self.DMA_ATTRIB_LAST)
self.wr_reg(self.HOST_BAR, self.HOST_DMA_ATTRIB + current_item_addr, attrib)
self.dma_item_cnt += 1
# Start DMA transfer
def start_dma(self):
self.dma_item_cnt = 0
self.dma_csr.wr_bit(self.R_DMA_CTL, self.DMA_CTL_START, 1)
# The following two lines should be removed
# when the GN4124 vhdl core will implement auto clear of start bit
#while(('Idle' == self.get_dma_status()) or
# ('Busy' == self.get_dma_status())):
# pass
self.dma_csr.wr_bit(self.R_DMA_CTL, self.DMA_CTL_START, 0)
# Abort DMA transfer
def abort_dma(self):
self.dma_item_cnt = 0
self.dma_csr.wr_bit(self.R_DMA_CTL, self.DMA_CTL_ABORT, 1)
# The following two lines should be removed
# when the GN4124 vhdl core will implement auto clear of start bit
while('Aborted' != self.get_dma_status()):
pass
self.dma_csr.wr_bit(self.R_DMA_CTL, self.DMA_CTL_ABORT, 0)
# Get memory page
def get_memory_page(self, page_nb):
data = []
for i in range(2**10):
data.append(self.rd_reg(self.HOST_BAR, (page_nb<<12)+(i<<2)))
return data
# Set memory page
def set_memory_page(self, page_nb, pattern):
for i in range(2**10):
self.wr_reg(self.HOST_BAR, (page_nb<<12)+(i<<2), pattern)
#!/usr/bin/python
import sys
import rr
import time
class DeviceNotFound(Exception):
def __init__(self, dev, addr):
self.dev = dev;
self.addr = addr;
def __str__(self):
return ("Device not found: %s at addr 0x%x" %(dev, addr))
class COpenCoresI2C:
# OpenCores I2C registers description
R_PREL = 0x0
R_PREH = 0x4
R_CTR = 0x8
R_TXR = 0xC
R_RXR = 0xC
R_CR = 0x10
R_SR = 0x10
CTR_EN = (1<<7)
CR_STA = (1<<7)
CR_STO = (1<<6)
CR_RD = (1<<5)
CR_WR = (1<<4)
CR_ACK = (1<<3)
SR_RXACK = (1<<7)
SR_TIP = (1<<1)
def wr_reg(self, addr, val):
self.bus.iwrite(0, self.base_addr + addr, 4, val)
def rd_reg(self,addr):
return self.bus.iread(0, self.base_addr + addr, 4)
# Function called during object creation
# bus = host bus (PCIe, VME, etc...)
# base_addr = I2C core base address
# prescaler = SCK prescaler, prescaler = (Fsys/(5*Fsck))-1
def __init__(self, bus, base_addr, prescaler):
self.bus = bus
self.base_addr = base_addr
self.wr_reg(self.R_CTR, 0)
#print("prescaler: %.4X") % prescaler
self.wr_reg(self.R_PREL, (prescaler & 0xff))
#print("PREL: %.2X") % self.rd_reg(self.R_PREL)
self.wr_reg(self.R_PREH, (prescaler >> 8))
#print("PREH: %.2X") % self.rd_reg(self.R_PREH)
self.wr_reg(self.R_CTR, self.CTR_EN)
#print("CTR: %.2X") % self.rd_reg(self.R_CTR)
if(not(self.rd_reg(self.R_CTR) & self.CTR_EN)):
print "Warning! I2C core is not enabled!"
def wait_busy(self):
while(self.rd_reg(self.R_SR) & self.SR_TIP):
pass
def start(self, addr, write_mode):
#print('i2c:start: addr=%.2X')%addr
addr = addr << 1
#print('i2c:start: addr=%.2X')%addr
if(write_mode == False):
addr = addr | 1
#print('i2c:start: addr=%.2X')%addr
self.wr_reg(self.R_TXR, addr)
#print("R_TXR: %.2X") % self.rd_reg(self.R_TXR)
self.wr_reg(self.R_CR, self.CR_STA | self.CR_WR)
self.wait_busy()
if(self.rd_reg(self.R_SR) & self.SR_RXACK):
raise DeviceNotFound("I2C", addr >> 1)
return "nack"
else:
return "ack"
def write(self, data, last):
self.wr_reg(self.R_TXR, data)
cmd = self.CR_WR
if(last):
cmd = cmd | self.CR_STO
self.wr_reg(self.R_CR, cmd)
self.wait_busy()
if(self.rd_reg(self.R_SR) & self.SR_RXACK):
raise I2CNotFound('No ACK upon write')
def read(self, last):
cmd = self.CR_RD
if(last):
cmd = cmd | self.CR_STO | self.CR_ACK
self.wr_reg(self.R_CR, cmd)
self.wait_busy()
return self.rd_reg(self.R_RXR)
def scan(self):
periph_addr = []
for i in range(0, 127):
addr = i << 1
# addr |= 1
self.wr_reg(self.R_TXR, addr)
self.wr_reg(self.R_CR, self.CR_STA | self.CR_WR)
self.wait_busy()
if(not(self.rd_reg(self.R_SR) & self.SR_RXACK)):
periph_addr.append(i)
print("Device found at address: 0x%.2X") % i
self.wr_reg(self.R_TXR, 0)
self.wr_reg(self.R_CR, self.CR_STO | self.CR_WR)
self.wait_busy()
return periph_addr
##########################################
# Usage example
#gennum = rr.Gennum();
#i2c = COpenCoresI2C(gennum, 0x80000, 500);
#!/usr/bin/python
import sys, termios, atexit
from select import select
# save the terminal settings
fd = sys.stdin.fileno()
new_term = termios.tcgetattr(fd)
old_term = termios.tcgetattr(fd)
# new terminal setting unbuffered
new_term[3] = (new_term[3] & ~termios.ICANON & ~termios.ECHO)
# switch to normal terminal
def set_normal_term():
termios.tcsetattr(fd, termios.TCSAFLUSH, old_term)
# switch to unbuffered terminal
def set_curses_term():
termios.tcsetattr(fd, termios.TCSAFLUSH, new_term)
def putch(ch):
sys.stdout.write(ch)
def getch():
return sys.stdin.read(1)
def getche():
ch = getch()
putch(ch)
return ch
def kbhit():
dr,dw,de = select([sys.stdin], [], [], 0)
return dr <> []
if __name__ == '__main__':
atexit.register(set_normal_term)
set_curses_term()
while 1:
if kbhit():
ch = getch()
break
sys.stdout.write('.')
print 'done'
#!/bin/bash
insmod ~/OHWR/wr-nic/software_test_dio_core/rawrabbit.ko vendor=0x10dc device=0x18d
#!/usr/bin/python
import sys
import rr
import time
class COpenCoresOneWire:
# OpenCores 1-wire registers description
R_CSR = 0x0
R_CDR = 0x4
CSR_DAT_MSK = (1<<0)
CSR_RST_MSK = (1<<1)
CSR_OVD_MSK = (1<<2)
CSR_CYC_MSK = (1<<3)
CSR_PWR_MSK = (1<<4)
CSR_IRQ_MSK = (1<<6)
CSR_IEN_MSK = (1<<7)
CSR_SEL_OFS = 8
CSR_SEL_MSK = (0xF<<8)
CSR_POWER_OFS = 16
CSR_POWER_MSK = (0xFFFF<<16)
CDR_NOR_MSK = (0xFFFF<<0)
CDR_OVD_OFS = 16
CDR_OVD_MSK = (0XFFFF<<16)
def wr_reg(self, addr, val):
self.bus.iwrite(0, self.base_addr + addr, 4, val)
def rd_reg(self,addr):
return self.bus.iread(0, self.base_addr + addr, 4)
# Function called during object creation
# bus = host bus (PCIe, VME, etc...)
# base_addr = 1-wire core base address
# clk_div_nor = clock divider normal operation, clk_div_nor = Fclk * 5E-6 - 1
# clk_div_ovd = clock divider overdrive operation, clk_div_ovd = Fclk * 1E-6 - 1
def __init__(self, bus, base_addr, clk_div_nor, clk_div_ovd):
self.bus = bus
self.base_addr = base_addr
#print('\n### Onewire class init ###')
#print("Clock divider (normal operation): %.4X") % clk_div_nor
#print("Clock divider (overdrive operation): %.4X") % clk_div_ovd
data = ((clk_div_nor & self.CDR_NOR_MSK) | ((clk_div_ovd<<self.CDR_OVD_OFS) & self.CDR_OVD_MSK))
#print('CRD register wr: %.8X') % data
self.wr_reg(self.R_CDR, data)
#print('CRD register rd: %.8X') % self.rd_reg(self.R_CDR)
# return: 1 -> presence pulse detected
# 0 -> no presence pulse detected
def reset(self, port):
data = ((port<<self.CSR_SEL_OFS) & self.CSR_SEL_MSK) | self.CSR_CYC_MSK | self.CSR_RST_MSK
#print('[onewire] Sending reset command, CSR: %.8X') % data
self.wr_reg(self.R_CSR, data)
while(self.rd_reg(self.R_CSR) & self.CSR_CYC_MSK):
pass
reg = self.rd_reg(self.R_CSR)
#print('[onewire] Reading CSR: %.8X') % reg
return ~reg & self.CSR_DAT_MSK
def slot(self, port, bit):
data = ((port<<self.CSR_SEL_OFS) & self.CSR_SEL_MSK) | self.CSR_CYC_MSK | (bit & self.CSR_DAT_MSK)
self.wr_reg(self.R_CSR, data)
while(self.rd_reg(self.R_CSR) & self.CSR_CYC_MSK):
pass
reg = self.rd_reg(self.R_CSR)
return reg & self.CSR_DAT_MSK
def read_bit(self, port):
return self.slot(port, 0x1)
def write_bit(self, port, bit):
return self.slot(port, bit)
def read_byte(self, port):
data = 0
for i in range(8):
data |= self.read_bit(port) << i
return data
def write_byte(self, port, byte):
data = 0
byte_old = byte
for i in range(8):
data |= self.write_bit(port, (byte & 0x1)) << i
byte >>= 1
if(byte_old == data):
return 0
else:
return -1
def write_block(self, port, block):
if(160 < len(block)):
return -1
data = []
for i in range(len(block)):
data.append(self.write_byte(port, block[i]))
return data
def read_block(self, port, length):
if(160 < length):
return -1
data = []
for i in range(length):
data.append(self.read_byte(port))
return data
#! /usr/bin/env python
# coding: utf8
class PtsException(Exception):
pass
class PtsCritical(PtsException):
"""critical error, abort the whole test suite"""
pass
class PtsError(PtsException):
"""error, continue remaining tests in test suite"""
pass
class PtsUser(PtsException):
"""error, user intervention required"""
pass
class PtsWarning(PtsException):
"""warning, a cautionary message should be displayed"""
pass
class PtsInvalid(PtsException):
"""reserved: invalid parameters"""
class PtsNoBatch(PtsInvalid):
"""reserved: a suite was created without batch of tests to run"""
pass
class PtsBadTestNo(PtsInvalid):
"""reserved: a bad test number was given"""
pass
if __name__ == '__main__':
pass
#! /usr/bin/env python
# :vi:ts=4 sw=4 et
from ctypes import *
import os, errno, re, sys, struct
import os.path
# python 2.4 kludge
if not 'SEEK_SET' in dir(os):
os.SEEK_SET = 0
# unsigned formats to unpack words
fmt = { 1: 'B', 2: 'H', 4: 'I', 8: 'L' }
# some defaults from rawrabbit.h
RR_DEVSEL_UNUSED = 0xffff
RR_DEFAULT_VENDOR = 0x1a39
RR_DEFAULT_DEVICE = 0x0004
RR_BAR_0 = 0x00000000
RR_BAR_2 = 0x20000000
RR_BAR_4 = 0x40000000
RR_BAR_BUF = 0xc0000000
bar_map = {
0 : RR_BAR_0,
2: RR_BAR_2,
4: RR_BAR_4,
0xc: RR_BAR_BUF }
# classes to interface with the driver via ctypes
Plist = c_int * 256
class RR_Devsel(Structure):
_fields_ = [
("vendor", c_ushort),
("device", c_ushort),
("subvendor", c_ushort),
("subdevice", c_ushort),
("bus", c_ushort),
("devfn", c_ushort),
]
class RR_U(Union):
_fields_ = [
("data8", c_ubyte),
("data16", c_ushort),
("data32", c_uint),
("data64", c_ulonglong),
]
class RR_Iocmd(Structure):
_anonymous_ = [ "data", ]
_fields_ = [
("address", c_uint),
("datasize", c_uint),
("data", RR_U),
]
def set_ld_library_path():
libpath = os.getenv('LD_LIBRARY_PATH')
here = os.getcwd()
libpath = here if not libpath else here + ':' + libpath
os.environ['LD_LIBRARY_PATH'] = libpath
class Gennum(object):
device = '/dev/rawrabbit'
rrlib = os.path.join(os.getcwd(), 'rrlib.so')
def __init__(self):
"""get a file descriptor for the Gennum device"""
set_ld_library_path()
self.lib = CDLL(Gennum.rrlib)
self.fd = os.open(Gennum.device, os.O_RDWR)
self.errno = 0
if self.fd < 0:
self.errno = self.fd
def iread(self, bar, offset, width):
"""do a read by means of the ioctl interface
bar = 0, 2, 4 (or c for DMA buffer access
offset = address within bar
width = data size (1, 2, 4 or 8 bytes)
"""
address = bar_map[bar] + offset
ds = RR_Iocmd(address=address, datasize=width)
self.errno = self.lib.rr_iread(self.fd, byref(ds))
return ds.data32
def read(self, bar, offset, width):
"""do a read by means of lseek+read
bar = 0, 2, 4 (or c for DMA buffer access
offset = address within bar
width = data size (1, 2, 4 or 8 bytes)
"""
address = bar_map[bar] + offset
self.errno = os.lseek(self.fd, address, os.SEEK_SET)
buf = os.read(self.fd, width)
return struct.unpack(fmt[width], buf)[0]
def iwrite(self, bar, offset, width, datum):
"""do a write by means of the ioctl interface
bar = 0, 2, 4 (or c for DMA buffer access
offset = address within bar
width = data size (1, 2, 4 or 8 bytes)
datum = value to be written
"""
address = bar_map[bar] + offset
ds = RR_Iocmd(address=address, datasize=width, data32=datum)
self.errno = self.lib.rr_iwrite(self.fd, byref(ds))
return ds.data32
def rd_reg(self, bar, addr):
return self.iread(bar, addr, 4)
def wr_reg(self, bar, addr, value):
# print("wr bar %d addr %x val %x" % (bar,addr,value))
self.iwrite(bar, addr, 4, value)
def write(self, bar, offset, width, datum):
"""do a write by means of lseek+write
bar = 0, 2, 4 (or c for DMA buffer access
offset = address within bar
width = data size (1, 2, 4 or 8 bytes)
datum = value to be written
"""
address = bar_map[bar] + offset
self.errno = os.lseek(self.fd, address, os.SEEK_SET)
return os.write(self.fd, struct.pack(fmt[width], datum))
def irqwait(self):
"""wait for an interrupt"""
return self.lib.rr_irqwait(self.fd);
def irqena(self):
"""enable the interrupt line"""
return self.lib.rr_irqena(self.fd);
def getdmasize(self):
"""return the size of the allocated DMA buffer (in bytes)"""
return self.lib.rr_getdmasize(self.fd);
def getplist(self):
"""get a list of pages for DMA access
The addresses returned, shifted by 12 bits, give the physical
addresses of the allocated pages
"""
plist = Plist()
self.lib.rr_getplist(self.fd, plist);
return plist
def info(self):
"""get a string describing the interface the driver is bound to
The syntax of the string is
vendor:device/dubvendor:subdevice@bus:devfn
"""
ds = RR_Devsel()
self.errno = self.lib.rr_devget(self.fd, byref(ds))
for key in RR_Devsel._fields_:
setattr(self, key[0], getattr(ds, key[0], RR_DEVSEL_UNUSED))
return '%04x:%04x/%04x:%04x@%04x:%04x' % (
ds.vendor, ds.device,
ds.subvendor, ds.subdevice,
ds.bus, ds.devfn)
def parse_addr(self, addr):
"""take a string of the form
vendor:device[/subvendor:subdevice][@bus:devfn]
and return a dictionary object with the corresponding values,
initialized to RR_DEVSEL_UNUSED when absent
"""
# address format
reg = ( r'(?i)^'
r'(?P<vendor>[a-f0-9]{1,4}):(?P<device>[a-f0-9]{1,4})'
r'(/(?P<subvendor>[a-f0-9]{1,4}):(?P<subdevice>[a-f0-9]{1,4}))?'
r'(@(?P<bus>[a-f0-9]{1,4}):(?P<devfn>[a-f0-9]{1,4}))?$' )
match = re.match(reg, addr).groupdict()
if not 'sub' in match:
match['subvendor'] = match['subdevice'] = RR_DEVSEL_UNUSED
if not 'geo' in match:
match['bus'] = match['devfn'] = RR_DEVSEL_UNUSED
for k, v in match.items():
if type(v) is str:
match[k] = int(v, 16)
return match
def bind(self, device):
"""bind the rawrabbit driver to a device
The device is specified with a syntax described in parse_addr
"""
d = self.parse_addr(device)
ds = RR_Devsel(**d)
self.errno = self.lib.rr_devsel(self.fd, byref(ds))
return self.errno
if __name__ == '__main__':
g = Gennum()
print g.parse_addr('1a39:0004/1a39:0004@0020:0000')
print g.bind('1a39:0004/1a39:0004@0020:0000')
print '%x' % g.write(bar=RR_BAR_4, offset=0xa08, width=4, datum=0xdeadface)
print '%x' % g.read(bar=RR_BAR_4, offset=0xa08, width=4)
print g.getdmasize()
for page in g.getplist():
print '%08x ' % (page<<12),
#!/usr/bin/python
#coding: utf8
from ptsexcept import *
from dio_fmc import *
import rr
import os
import sys
import kbhit
"""
test01: Tests all the outputs and the FMC front panel LEDs
"""
def pause():
raw_input("press key\n")
def print_lemos(dio):
for lemon in range(5):
print dio.get_in(lemon),
def osc_lemos(dio):
while(not kbhit.kbhit()):
for lemon in range(5):
dio.set_out(lemon,0)
time.sleep(0.05)
dio.set_out(0,1)
time.sleep(0.05)
for lemon in range(5):
dio.set_out(lemon,0)
if lemon < 5:
dio.set_out(lemon+1,1)
time.sleep(0.05)
for lemon in range(5):
dio.set_out(lemon,0)
def main(default_directory="."):
spec = rr.Gennum()
dio = CFmcDio(spec, 0x80000)
#dio.set_out(0,1)
#dio.set_dir(0, 1)
#dio.set_term(0, 0)
#exit()
print "(3 devices expected)"
print
print "FMC temperature: {:3}°C".format(dio.get_temp())
print "(expected room or computer temperature)"
print
print "Flashing board LEDs. Press key"
while(not kbhit.kbhit()):
for nled in range(2):
dio.set_led(nled,1)
time.sleep(0.2)
print ".",
for nled in range(2):
dio.set_led(nled,0)
time.sleep(0.2)
sys.stdout.flush()
kbhit.getch()
print
print "Oscillating LEMOs without load. Press key"
print "(2.06V-amplitude pulses expected in all ports)"
for lemon in range(5):
dio.set_dir(lemon, 1)
dio.set_term(lemon, 0)
osc_lemos(dio)
kbhit.getch()
print "Oscillating LEMOs with load. Press key"
print "(1.58V-amplitude pulses expected in all ports)"
for lemon in range(5):
dio.set_term(lemon, 1)
osc_lemos(dio)
for lemon in range(5):
dio.set_term(lemon, 0)
kbhit.getch()
print "Oscillating LEMOs with driver disabled. Press key"
print "(no pulse expected in all ports)"
for lemon in range(5):
dio.set_dir(lemon, 0)
osc_lemos(dio)
kbhit.getch()
for lemon in range(5):
dio.set_in_threshold(lemon,10)
print "Input threshold set to an intermediate level ({}).".format(dio.get_in_threshold(0))
print "Value of LEMOs with all drivers and terminations disabled:",
for lemon in range(5):
dio.set_term(lemon, 0)
dio.set_dir(lemon, 0)
dio.set_out(lemon, 1)
print_lemos(dio)
for lemon in range(5):
dio.set_out(lemon, 0)
print
print "(zero vector expected)"
print
print "Terminations off"
for lemon in range(5):
dio.set_dir(lemon, 1)
dio.set_out(lemon, 1)
print "driver",lemon,"on. INPUT:",
print_lemos(dio)
print
dio.set_dir(lemon, 0)
dio.set_out(lemon, 0)
print "(identity matrix expected)"
print
print "Terminations on"
for lemon in range(5):
dio.set_dir(lemon, 1)
dio.set_out(lemon, 1)
dio.set_term(lemon, 1)
print "driver",lemon,"on. INPUT:",
print_lemos(dio)
print
dio.set_dir(lemon, 0)
dio.set_out(lemon, 0)
dio.set_term(lemon, 0)
print "(zero matrix expected)"
print
print "EEPROM sequence written. DATA:",
oldedatas=dio.eeprom.rd_data(0,20)
time.sleep(0.1)
dio.eeprom.wr_data(0,range(20))
time.sleep(0.1)
edatas=dio.eeprom.rd_data(0,20)
for edata in edatas:
print edata,
print
time.sleep(0.1)
# oldedatas=[]
# oldedatas.extend(255 for number in range(30))
dio.eeprom.wr_data(0,oldedatas)
print "(0 to 19 sequence expected)"
print
for lemon in range(5):
print "Testing LEMO:",lemon
dio.set_dir(lemon, 1)
dio.set_out(lemon, 1)
dio.set_term(lemon, 1)
for odac in range(0,256,32):
dio.set_in_threshold(lemon,odac)
time.sleep(0.005)
print "DAC Out: {:3} LEMOs:".format(odac),
print_lemos(dio)
print
dio.set_dir(lemon, 0)
dio.set_out(lemon, 0)
dio.set_term(lemon, 0)
print "(the first 4 elements in column",lemon+1,"are expected to be 1)"
print
exit()
"""
ask = ""
## Set all LEMOS output to 2V
lemos = 0, 1, 2, 3, 4
for k in lemos:
dio.set_out(k,1)
ask = ""
while ((ask != "Y") and (ask != "N")) :
print "-------------------------------------------------------------"
print "\t Testing Outputs "
ask = raw_input("Does the Output is 2V? [Y/N]")
ask = ask.upper()
print "-------------------------------------------------------------"
if (ask == "N") :
raise PtsError("Error loading FW through the Flash memory or there is a problem with the LEDs")
## Set all LEMOS output to 0V
for k in lemos:
dio.set_out(k,0)
ask = ""
while ((ask != "Y") and (ask != "N")) :
print "-------------------------------------------------------------"
print "\t Testing LEDs "
ask = raw_input("Does the Output is 0V? [Y/N]")
ask = ask.upper()
print "-------------------------------------------------------------"
if (ask == "N") :
raise PtsError("Error loading FW through the Flash memory or there is a problem with the LEDs")
"""
if __name__ == "__main__":
main(".")
#!/usr/bin/python
#coding: utf8
from ptsexcept import *
from dio_fmc import *
import rr
import os
import sys
import gn4124
import kbhit
"""
Tests DIO core (GEN AND STAMPER PULSE)
"""
def pause():
raw_input("press key\n")
GN4124_CSR = 0x0
def main(default_directory="."):
print "(-------------STARTING TEST-----------------)"
# Configure the FPGA using the program fpga_loader
path_fpga_loader = './fpga_loader'
path_firmware = '../syn/spec/wr_nic_top.bin'
firmware_loader = os.path.join(default_directory, path_fpga_loader)
bitstream = os.path.join(default_directory, path_firmware)
print "Loading firmware: %s" % (firmware_loader + ' ' + bitstream)
os.system( firmware_loader + ' ' + bitstream )
# Load board library and open the corresponding device
spec = rr.Gennum()
gennum = gn4124.CGN4124(spec, GN4124_CSR)
dio = CFmcDio(spec, 0x62000)
print "(3 devices expected)"
print
print ("FMC temperature: %3.3f°C" % dio.get_temp())
print "(expected room or computer temperature)"
print
# Lemo output configuration
print "(------------CONFIGURING DIO CHANNELS--------------)"
print "Value of LEMOs with all drivers enabled and terminations disabled"
for lemon in range(5):
dio.set_term(lemon, 0)
dio.set_dir(lemon, 1) # enable output
dio.set_in_threshold(lemon,15)
print "Input threshold set to an intermediate level ({}).".format(dio.get_in_threshold(0))
print
print "(------------CONFIGURING INTERRUPTS--------------)"
# DIO Interrupts
print "(DIO Interrupts)"
#dio.set_reg(0x64, 0x1f) # Interrupts when the fifos have datas (UTC time from the pulse stamper)
dio.set_reg(0x64, 0x3ff) # fifos and pulse gen rdy interrupts
mask_irq = dio.get_reg(0x68)
print "MASK IRQ DIO =>", mask_irq
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq
# VIC Interrupts
print "(VIC Interrupts)"
VIC = VIC_irq(spec, 0x60000)
VIC.set_reg(0x0, 0x3) # control register
control_irq_vic = VIC.get_reg(0x0)
print "CONTROL IRQ VIC =>", control_irq_vic
VIC.set_reg(0x8, 0x7) # enable register
mask_irq_vic = VIC.get_reg(0x10)
print "MASK IRQ VIC =>", mask_irq_vic
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
# Simulate interrupts
#VIC.set_reg(0x18, 0x7) # generate soft irq
#status_irq_vic = VIC.get_reg(0x4)
#print "STATUS VIC IRQ =>", status_irq_vic
# Checking interrupts at pc level
#print "Waiting irq ..."
#spec.irqena()
#gennum.set_interrupt_config()
#a=gennum.wait_irq()
#b=spec.irqwait()
#print "Is this working?", a,b
# Read WRCORE time --> HANG UP THE PC!!!!!!!
#WR_CORE_AD =0x0
#PRI_CROSSBAR_AD=0x00020000 # Second bridge
#SEC_CROSSBAR_AD=0x00000300 # PPS
#time = wrcore_time(spec, (WR_CORE_AD + PRI_CROSSBAR_AD+SEC_CROSSBAR_AD))
#print "eso", time
#seconds= time.get_reg(0x8) NNOOOOOOO!!!!!! 0x00020308
#cycles = time.get_reg(0x4)
#seconds=10
#cycles=0
#print "Time is: ", seconds, "and", cycles*8, "(cycles", cycles, ")"
###################################################################################
# START LEMO OUTPUT DANCING ...
print
print "(------------START LEMO CONFIGURATION--------------)"
#print "Note: The dummy time core is already running after configuring the fpga, therefore you should run this program"
#print "before the configured trigger time below."
# BASIC GPIO FUNCTIONALITY TEST
dio.set_reg(0x3C, 0x00) # channels as GPIOs
for lemon in range(5):
dio.set_dir(lemon, 1) # enable output
dio.set_out(lemon,0)
val=dio.get_in (lemon)
print "Set value for channel", lemon, " to 0, read: ", val
dio.set_out(lemon,1)
val=dio.get_in (lemon)
print "Set value for channel", lemon, " to 1, read: ", val
# Please connect channel 0 and 1 with a lemo wire for this test
dio.set_dir(0, 0) # enable channel 1 output and dissble channel 0
dio.set_dir(1,1)
dio.set_out(1,0)
val=dio.get_in (0)
print "Channel 1 write 0, channel 0 read", val
dio.set_out(1,1)
val=dio.get_in (0)
print "Channel 1 write 1, channel 0 read", val
# Osciloscope test
#dio.set_out(4,1)
#pause()
#dio.set_out(4,0)
#pause()
#dio.set_out(4,1)
#pause()
#dio.set_out(4,0)
#pause()
# TIME-LENGTH PROGRAMMABLE PULSES TESTS
dio.set_reg(0x3C, 0x1f) #Generate a programmable/immediate pulse of different length
# Setting pulse length
dio.set_reg(0x48, 0x1)
dio.set_reg(0x4C, 0x8)
dio.set_reg(0x50, 0x10)
dio.set_reg(0x54, 0x20)
dio.set_reg(0x58, 0x40)
# Time-stamps FIFOs registers address
adr_status_fifo = [0x7c, 0x8c, 0x9c, 0xac, 0xbc]
adr_time = [0x70, 0x80, 0x90, 0xa0, 0xb0] # we just take the lowest 32 bits for the seconds
adr_cycles = [0x78, 0x88, 0x98, 0xa8, 0xb8]
# Flushng fifos from previous game
print
for dio_pulse in range(5): # reading of pulses
print "------Flushing fifos ", dio_pulse, ":"
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = 0
while((status_fifo_reg & 0xff) != 0x0):
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "Pulse ", cont
print "Seconds DIO dio_pulse =>", time
print "cycles DIO dio_pulse =>", cycles
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = cont + 1
print
###################################################################################################
# Inmmediate output test
print
print "(-----------IMMEDIATE PULSE GENERATION (5 pulses for each DIO)---------------)"
#pause()
for num_pulses in range(5):
dio.set_reg(0x5C, 0x1f) #Generate a pulse of different length
#status_irq = dio.get_reg(0x6c)
##print "STATUS IRQ DIO =>", status_irq
print "5 Immediate pulse of different length has been generated for each channel"
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq # 31 means the 5 fifos has data
print
for dio_pulse in range(5): # reading of pulses
print "------Pulses from DIO ", dio_pulse, ":"
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = 0
while((status_fifo_reg & 0xff) != 0x0):
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "Pulse ", cont
print "Seconds DIO dio_pulse =>", time
print "cycles DIO dio_pulse =>", cycles
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = cont + 1
print
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq # we have read the fifo, no data to read => 0
############################################################################
print
print "Starting time-programmable test"
print
print "(------------CONFIGURING TRIGGER TIME FOR EACH LEMO--------------)"
# Programmable output test
dio_rdy = dio.get_reg(0x44)
if dio_rdy != 0x1f:
print "Some pulse_gen module isn't ready to accept next trigger time tag, dio_rdy:", dio_rdy
exit()
#TIME TRIGGERS VALUES FOR EACH CHANNEL
dio.set_reg(0x0, 3) #trig0 seconds_low
dio.set_reg(0x4, 0) #trig0 seconds_high
dio.set_reg(0x8, 0) #trig0 cycles
dio.set_reg(0xc, 10) #trig1 seconds_low
dio.set_reg(0x10, 0) #trig1 seconds_high
dio.set_reg(0x14, 67) #trig1 cycles
dio.set_reg(0x18, 15) #trig2 seconds_low
dio.set_reg(0x1c, 0) #trig2 seconds_high
dio.set_reg(0x20, 430) #trig2 cycles
dio.set_reg(0x24, 20) #trig3 seconds_low
dio.set_reg(0x28, 0) #trig3 seconds_high
dio.set_reg(0x2c, 94) #trig3 cycles
dio.set_reg(0x30, 25) #trig4 seconds_low
dio.set_reg(0x34, 0) #trig4 seconds_high
dio.set_reg(0x38, 98) #trig4 cycles
dio.set_reg(0x40, 0x1f) # channell x trigger strobe
print
for dio_pulse in range(5):
print "Waiting pulse stamper from DIO ", dio_pulse," ...."
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
while((status_fifo_reg & 0xff) != 0x1): #counter = 0x1
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "DIO dio_pulse seconds =>", time
print "DIO dio_pulse cycles =>", cycles
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
print
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq
exit()
if __name__ == "__main__":
main(".")
#!/usr/bin/python
#coding: utf8
from ptsexcept import *
from dio_fmc import *
import rr
import os
import sys
import gn4124
import kbhit
"""
Tests DIO core (GEN AND STAMPER PULSE)
"""
def pause():
raw_input("press key\n")
GN4124_CSR = 0x0
def main(default_directory="."):
print "(-------------STARTING TEST-----------------)"
# Configure the FPGA using the program fpga_loader
path_fpga_loader = './fpga_loader'
path_firmware = '../syn/spec/wr_nic_top.bin'
firmware_loader = os.path.join(default_directory, path_fpga_loader)
bitstream = os.path.join(default_directory, path_firmware)
print "Loading firmware: %s" % (firmware_loader + ' ' + bitstream)
os.system( firmware_loader + ' ' + bitstream )
# Load board library and open the corresponding device
spec = rr.Gennum()
gennum = gn4124.CGN4124(spec, GN4124_CSR)
dio = CFmcDio(spec, 0x62000)
print "(3 devices expected)"
print
print ("FMC temperature: %3.3f°C" % dio.get_temp())
print "(expected room or computer temperature)"
print
# Lemo output configuration
print "(------------CONFIGURING DIO CHANNELS--------------)"
print "Value of LEMOs with all drivers enabled and terminations disabled"
for lemon in range(5):
dio.set_term(lemon, 0)
dio.set_dir(lemon, 1) # enable output
dio.set_in_threshold(lemon,15)
print "Input threshold set to an intermediate level ({}).".format(dio.get_in_threshold(0))
print
print "(------------CONFIGURING INTERRUPTS--------------)"
# DIO Interrupts
print "(DIO Interrupts)"
dio.set_reg(0x64, 0x1f) # Interrupts when the fifos have datas (UTC time from the pulse stamper)
#dio.set_reg(0x64, 0x3ff) # fifos and pulse gen rdy interrupts
mask_irq = dio.get_reg(0x68)
print "MASK IRQ DIO =>", mask_irq
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq
# VIC Interrupts
print "(VIC Interrupts)"
VIC = VIC_irq(spec, 0x60000)
VIC.set_reg(0x0, 0x3) # control register
control_irq_vic = VIC.get_reg(0x0)
print "CONTROL IRQ VIC =>", control_irq_vic
VIC.set_reg(0x8, 0x7) # enable register
mask_irq_vic = VIC.get_reg(0x10)
print "MASK IRQ VIC =>", mask_irq_vic
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
# Simulate interrupts
#VIC.set_reg(0x18, 0x7) # generate soft irq
#status_irq_vic = VIC.get_reg(0x4)
#print "STATUS VIC IRQ AFTER SOFTWARE INTERRUPTS =>", status_irq_vic
# Checking interrupts at pc level
#print "Waiting irq ..."
#spec.irqena()
#gennum.set_interrupt_config()
#a=gennum.wait_irq()
#b=spec.irqwait()
#print "Is this working?", a,b
# Read WRCORE time --> HANG UP THE PC!!!!!!!
#WR_CORE_AD =0x0
#PRI_CROSSBAR_AD=0x00020000 # Second bridge
#SEC_CROSSBAR_AD=0x00000300 # PPS
#time = wrcore_time(spec, (WR_CORE_AD + PRI_CROSSBAR_AD+SEC_CROSSBAR_AD))
#print "eso", time
#seconds= time.get_reg(0x8) NNOOOOOOO!!!!!! 0x00020308
#cycles = time.get_reg(0x4)
#seconds=10
#cycles=0
#print "Time is: ", seconds, "and", cycles*8, "(cycles", cycles, ")"
###################################################################################
# START LEMO OUTPUT DANCING ...
print
print "(------------START LEMO CONFIGURATION--------------)"
#print "Note: The dummy time core is already running after configuring the fpga, therefore you should run this program"
#print "before the configured trigger time below."
# BASIC GPIO FUNCTIONALITY TEST
dio.set_reg(0x3C, 0x00) # channels as GPIOs
for lemon in range(5):
dio.set_dir(lemon, 1) # enable output
dio.set_out(lemon,0)
val=dio.get_in (lemon)
print "Set value for channel", lemon, " to 0, read: ", val
dio.set_out(lemon,1)
val=dio.get_in (lemon)
print "Set value for channel", lemon, " to 1, read: ", val
# Please connect channel 0 and 1 with a lemo wire for this test
dio.set_dir(0, 0) # enable channel 1 output and dissble channel 0
dio.set_dir(1,1)
dio.set_out(1,0)
val=dio.get_in (0)
print "Channel 1 write 0, channel 0 read", val
dio.set_out(1,1)
val=dio.get_in (0)
print "Channel 1 write 1, channel 0 read", val
# Osciloscope test
#dio.set_out(4,1)
#pause()
#dio.set_out(4,0)
#pause()
#dio.set_out(4,1)
#pause()
#dio.set_out(4,0)
#pause()
# TIME-LENGTH PROGRAMMABLE PULSES TESTS
dio.set_reg(0x3C, 0x1f) #Generate a programmable/immediate pulse of different length
# Setting pulse length
dio.set_reg(0x48, 0x1)
dio.set_reg(0x4C, 0x8)
dio.set_reg(0x50, 0x10)
dio.set_reg(0x54, 0x1)
dio.set_reg(0x58, 0x40)
# Time-stamps FIFOs registers address
adr_status_fifo = [0x7c, 0x8c, 0x9c, 0xac, 0xbc]
adr_time = [0x70, 0x80, 0x90, 0xa0, 0xb0] # we just take the lowest 32 bits for the seconds
adr_cycles = [0x78, 0x88, 0x98, 0xa8, 0xb8]
# Flushng fifos from previous game
print
for dio_pulse in range(5): # reading of pulses
print "------Flushing fifos ", dio_pulse, ":"
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = 0
while((status_fifo_reg & 0xff) != 0x0):
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "Pulse ", cont
print "Seconds DIO dio_pulse =>", time
print "cycles DIO dio_pulse =>", cycles
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = cont + 1
print
###################################################################################################
# Inmmediate output test
print
print "(-----------IMMEDIATE PULSE GENERATION (5 pulses for each DIO)---------------)"
#pause()
for num_pulses in range(5):
dio.set_reg(0x5C, 0x1f) #Generate a pulse of different length
#status_irq = dio.get_reg(0x6c)
##print "STATUS IRQ DIO =>", status_irq
print "5 Immediate pulse of different length has been generated for each channel"
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq # 31 means the 5 fifos has data
print
for dio_pulse in range(5): # reading of pulses
print "------Pulses from DIO ", dio_pulse, ":"
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = 0
while((status_fifo_reg & 0xff) != 0x0):
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "Pulse ", cont
print "Seconds DIO dio_pulse =>", time
print "cycles DIO dio_pulse =>", cycles
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
cont = cont + 1
print
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq # we have read the fifo, no data to read => 0
############################################################################
print
print "Starting time-programmable test"
print
print "(------------TIME-TRIGGER BASED OUTPUTS--------------)"
# Programmable output test
dio_rdy = dio.get_reg(0x44)
if dio_rdy != 0x1f:
print "Some pulse_gen module isn't ready to accept next trigger time tag, dio_rdy:", dio_rdy
exit()
#TIME TRIGGERS VALUES FOR EACH CHANNEL
dio.set_reg(0x0, 3) #trig0 seconds_low
dio.set_reg(0x4, 0) #trig0 seconds_high
dio.set_reg(0x8, 0) #trig0 cycles
dio.set_reg(0xc, 5) #trig1 seconds_low
dio.set_reg(0x10, 0) #trig1 seconds_high
dio.set_reg(0x14, 67) #trig1 cycles
dio.set_reg(0x18, 10) #trig2 seconds_low
dio.set_reg(0x1c, 0) #trig2 seconds_high
dio.set_reg(0x20, 430) #trig2 cycles
dio.set_reg(0x24, 12) #trig3 seconds_low
dio.set_reg(0x28, 0) #trig3 seconds_high
dio.set_reg(0x2c, 94) #trig3 cycles
dio.set_reg(0x30, 15) #trig4 seconds_low
dio.set_reg(0x34, 0) #trig4 seconds_high
dio.set_reg(0x38, 98) #trig4 cycles
dio.set_reg(0x40, 0x1f) # channell x trigger strobe
print
for dio_pulse in range(5):
print "Waiting pulse stamper from DIO ", dio_pulse," ...."
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
while((status_fifo_reg & 0xff) != 0x1): #counter = 0x1
status_fifo_reg = dio.get_reg(adr_status_fifo[dio_pulse])
time = dio.get_reg_long(adr_time[dio_pulse])
cycles = dio.get_reg(adr_cycles[dio_pulse])
print "DIO dio_pulse seconds =>", time
print "DIO dio_pulse cycles =>", cycles
print
status_irq_vic = VIC.get_reg(0x4)
print "STATUS VIC IRQ =>", status_irq_vic
print
status_irq = dio.get_reg(0x6c)
print "STATUS IRQ DIO =>", status_irq
exit()
if __name__ == "__main__":
main(".")
#!/usr/bin/python
#simple driver for (x)wb_gpio_port from general-cores library
class CGPIO:
def __init__(self, bus, base):
self.bus =bus
self.base = base
#returns the value on pin pin
def inp(self, pin):
bank= 0x20 * (pin / 32);
r = self.bus.rd_reg(0, self.base + bank + 0xc);
return (r >> pin) & 1;
#sets the pin "pin" to value "val"
def outp(self, pin, val):
bank= 0x20 * (pin >> 5);
if(val):
self.bus.wr_reg(0, self.base + bank + 0x4, (1<<(pin&0x1f)))
else:
self.bus.wr_reg(0, self.base + bank + 0x0, (1<<(pin&0x1f)))
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment