"""
from random import randint
from math import ceil, floor
-from nmigen import Elaboratable, Module, Signal, Record, Array, Cat
+from nmigen import Elaboratable, Module, Signal, Record, Array, Cat, Const
from nmigen.hdl.rec import Layout
from nmigen.utils import log2_int
from nmigen.cli import rtlil
class SimpleGPIO(Elaboratable):
def __init__(self, wordsize=4, n_gpio=16):
- print("SimpleGPIO: WB Data # of bytes: {0}, # of GPIOs: {1}"
- .format(wordsize, n_gpio))
self.wordsize = wordsize
self.n_gpio = n_gpio
+ self.n_rows = ceil(self.n_gpio / self.wordsize)
+ print("SimpleGPIO: WB Data # of bytes: {0}, #GPIOs: {1}, Rows: {2}"
+ .format(self.wordsize, self.n_gpio, self.n_rows))
class Spec: pass
spec = Spec()
spec.addr_wid = 30
spec.reg_wid = wordsize*8 # 32
self.bus = Record(make_wb_layout(spec), name="gpio_wb")
- print("CSRBUS layout: ", csrbus_layout)
- # create array - probably a cleaner way to do this...
+ #print("CSRBUS layout: ", csrbus_layout)
+ # MultiCSR read and write buses
+ temp = []
+ for i in range(self.wordsize):
+ temp_str = "rd_word{}".format(i)
+ temp.append(Record(name=temp_str, layout=csrbus_layout))
+ self.rd_multicsr = Array(temp)
+
temp = []
for i in range(self.wordsize):
temp_str = "word{}".format(i)
bus = self.bus
wb_rd_data = bus.dat_r
wb_wr_data = bus.dat_w
+ wb_rd_data_reg = Signal(self.wordsize*8) # same len as WB bus
wb_ack = bus.ack
gpio_ports = self.gpio_ports
multi = self.multicsrbus
+ rd_multi = self.rd_multicsr
- comb += wb_ack.eq(0)
-
- row_start = Signal(log2_int(self.n_gpio))
# Flag for indicating rd/wr transactions
new_transaction = Signal(1)
# One address used to configure CSR, set output, read input
with m.If(bus.cyc & bus.stb):
- comb += wb_ack.eq(1) # always ack
- # Probably wasteful
- sync += row_start.eq(bus.adr * self.wordsize)
+ sync += wb_ack.eq(1) # always ack, always delayed
+ # TODO: is this needed anymore?
sync += new_transaction.eq(1)
+ # Concatinate the GPIO configs that are on the same "row" or
+ # address and send
+ multi_cat = []
+ for i in range(0, self.wordsize):
+ multi_cat.append(rd_multi[i])
+ sync += wb_rd_data_reg.eq(Cat(multi_cat))
with m.If(bus.we): # write
# Configure CSR
for byte in range(0, self.wordsize):
+ # TODO: wasteful... convert to Cat(), somehow
sync += multi[byte].eq(wb_wr_data[byte*8:8+byte*8])
- with m.Else(): # read
- # Concatinate the GPIO configs that are on the same "row" or
- # address and send
- multi_cat = []
- for i in range(0, self.wordsize):
- multi_cat.append(multi[i])
- comb += wb_rd_data.eq(Cat(multi_cat))
+ with m.Elif(wb_ack): # read (and acked)
+ comb += wb_rd_data.eq(wb_rd_data_reg)
with m.Else():
sync += new_transaction.eq(0)
- # Update the state of "io" while no WB transactions
- for byte in range(0, self.wordsize):
- with m.If(gpio_ports[row_start+byte].oe):
- sync += multi[byte].io.eq(gpio_ports[row_start+byte].o)
- with m.Else():
- sync += multi[byte].io.eq(gpio_ports[row_start+byte].i)
+ sync += wb_ack.eq(0)
+
# Only update GPIOs config if a new transaction happened last cycle
# (read or write). Always lags from multi csrbus by 1 clk cycle, most
# sane way I could think of while using Record().
with m.If(new_transaction):
- for byte in range(0, self.wordsize):
- sync += gpio_ports[row_start+byte].oe.eq(multi[byte].oe)
- sync += gpio_ports[row_start+byte].puen.eq(multi[byte].puen)
- sync += gpio_ports[row_start+byte].pden.eq(multi[byte].pden)
- # Check to prevent output being set if GPIO configured as input
- # TODO: No checking is done if ie/oe high together
- with m.If(gpio_ports[row_start+byte].oe):
- sync += gpio_ports[row_start+byte].o.eq(multi[byte].io)
- sync += gpio_ports[row_start+byte].bank.eq(multi[byte].bank)
+ self.connect_wr_bus_to_gpio(m, sync, bus.adr, gpio_ports, multi)
+ self.connect_gpio_to_rd_bus(m, sync, bus.adr, gpio_ports, rd_multi)
+ #with m.Else():
+ # print("Copy gpio_ports to multi...")
+ # sync += multi[]
return m
+ def connect_wr_bus_to_gpio(self, module, domain, addr, gp, multi):
+ if self.n_gpio > self.wordsize:
+ print("#GPIOs is greater than, and is a multiple of WB wordsize")
+ # Case where all gpios fit within full words
+ if self.n_gpio % self.wordsize == 0:
+ for row in range(self.n_rows):
+ with module.If(addr == Const(row)):
+ offset = row*self.wordsize
+ for byte in range(self.wordsize):
+ domain += gp[byte+offset].oe.eq(multi[byte].oe)
+ domain += gp[byte+offset].puen.eq(multi[byte].puen)
+ domain += gp[byte+offset].pden.eq(multi[byte].pden)
+ # prevent output being set if GPIO configured as i
+ # TODO: No checking is done if ie/oe high together
+ with module.If(multi[byte].oe):
+ domain += gp[byte+offset].o.eq(multi[byte].io)
+ with module.Else():
+ domain += multi[byte].io.eq(gp[byte+offset].i)
+ domain += gp[byte+offset].bank.eq(multi[byte].bank)
+ else:
+ # TODO: This is a complex case, not needed atm
+ print("#GPIOs is greater than WB wordsize")
+ print("But not fully fitting in words...")
+ print("NOT IMPLEMENTED THIS CASE")
+ raise
+ else:
+ print("#GPIOs is less or equal to WB wordsize (in bytes)")
+ for byte in range(self.n_gpio):
+ domain += gp[byte].oe.eq(multi[byte].oe)
+ domain += gp[byte].puen.eq(multi[byte].puen)
+ domain += gp[byte].pden.eq(multi[byte].pden)
+ # Check to prevent output being set if GPIO configured as i
+ # TODO: No checking is done if ie/oe high together
+ with module.If(multi[byte].oe):
+ domain += gp[byte].o.eq(multi[byte].io)
+ with module.Else():
+ domain += gp[byte].o.eq(0)
+ domain += gp[byte].bank.eq(multi[byte].bank)
+
+ def connect_gpio_to_rd_bus(self, module, domain, addr, gp, multi):
+ if self.n_gpio > self.wordsize:
+ print("#GPIOs is greater than, and is a multiple of WB wordsize")
+ # Case where all gpios fit within full words
+ if self.n_gpio % self.wordsize == 0:
+ for row in range(self.n_rows):
+ with module.If(addr == Const(row)):
+ offset = row*self.wordsize
+ for byte in range(self.wordsize):
+ domain += multi[byte].oe.eq(gp[byte+offset].oe)
+ domain += multi[byte].puen.eq(gp[byte+offset].puen)
+ domain += multi[byte].pden.eq(gp[byte+offset].pden)
+ # prevent output being set if GPIO configured as i
+ # TODO: No checking is done if ie/oe high together
+ with module.If(gp[byte+offset].oe):
+ domain += multi[byte].io.eq(gp[byte+offset].o)
+ with module.Else():
+ domain += multi[byte].io.eq(gp[byte+offset].i)
+ domain += multi[byte].bank.eq(gp[byte+offset].bank)
+ else:
+ # TODO: This is a complex case, not needed atm
+ print("#GPIOs is greater than WB wordsize")
+ print("NOT IMPLEMENTED THIS CASE")
+ raise
+ else:
+ print("#GPIOs is less or equal to WB wordsize (in bytes)")
+ for byte in range(self.n_gpio):
+ domain += multi[byte].oe.eq(gp[gpio].oe)
+ domain += multi[byte].puen.eq(gp[gpio].puen)
+ domain += multi[byte].pden.eq(gp[gpio].pden)
+ # Check to prevent output being set if GPIO configured as i
+ # TODO: No checking is done if ie/oe high together
+ with module.If(multi[byte].oe):
+ domain += multi[byte].io.eq(gp[gpio].o)
+ with module.Else():
+ domain += multi[byte].io.eq(gp[byte].i)
+ domain += multi[byte].bank.eq(gp[gpio].bank)
+
+
def __iter__(self):
for field in self.bus.fields.values():
yield field
- #yield self.gpio_o
+ for gpio in range(len(self.gpio_ports)):
+ for field in self.gpio_ports[gpio].fields.values():
+ yield field
def ports(self):
return list(self)
+"""
def gpio_test_in_pattern(dut, pattern):
num_gpios = len(dut.gpio_ports)
print("Test pattern:")
print(pattern)
for pat in range(0, len(pattern)):
for gpio in range(0, num_gpios):
- yield from gpio_set_in_pad(dut, gpio, pattern[pat])
+ yield gpio_set_in_pad(dut, gpio, pattern[pat])
yield
temp = yield from gpio_rd_input(dut, gpio)
print("Pattern: {0}, Reading {1}".format(pattern[pat], temp))
pat += 1
if pat == len(pattern):
break
+"""
def test_gpio_single(dut, gpio, use_random=True):
oe = 1
# Object for storing each gpio's config state
class GPIOManager():
- def __init__(self, dut, layout):
+ def __init__(self, dut, layout, wb_bus):
self.dut = dut
+ self.wb_bus = wb_bus
# arrangement of config bits making up csr word
self.csr_layout = layout
self.shift_dict = self._create_shift_dict()
return oe, ie, puen, pden, io, bank
def rd_csr(self, row_start):
- row_word = yield from wb_read(self.dut.bus, row_start)
+ row_word = yield from wb_read(self.wb_bus, row_start)
print("Returned CSR: {0:x}".format(row_word))
return row_word
curr_gpio = row_addr * self.wordsize
config_word = 0
for byte in range(0, self.wordsize):
- if curr_gpio > self.n_gpios:
+ if curr_gpio >= self.n_gpios:
break
config_word += self.shadow_csr[curr_gpio].packed << (8 * byte)
#print("Reading GPIO{} shadow reg".format(curr_gpio))
curr_gpio += 1
print("Writing shadow CSRs val {0:x} to row addr {1:x}"
.format(config_word, row_addr))
- yield from wb_write(self.dut.bus, row_addr, config_word)
+ yield from wb_write(self.wb_bus, row_addr, config_word)
yield # Allow one clk cycle to propagate
if(check):
curr_gpio = row_addr * self.wordsize
single_csr = 0
for byte in range(0, self.wordsize):
- if curr_gpio > self.n_gpios:
+ if curr_gpio >= self.n_gpios:
break
single_csr = (read_word >> (8 * byte)) & 0xFF
#print("Updating GPIO{0} shadow reg to {1:x}"
gpio_internal_traces = ('Internal', [
('clk', 'in'),
('new_transaction'),
- ('row_start[2:0]'),
('rst', 'in')
])
traces.append(gpio_internal_traces)
def test_gpioman(dut):
print("------START----------------------")
- gpios = GPIOManager(dut, csrbus_layout)
+ gpios = GPIOManager(dut, csrbus_layout, dut.bus)
gpios.print_info()
#gpios._parse_gpio_arg("all")
#gpios._parse_gpio_arg("0")
- gpios._parse_gpio_arg("1-3")
+ #gpios._parse_gpio_arg("1-3")
#gpios._parse_gpio_arg("20")
oe = 1
bank = 3
yield from gpios.config("0-3", oe=1, ie=0, puen=0, pden=1, outval=0, bank=2)
ie = 1
- yield from gpios.config("4-7", oe=0, ie=1, puen=0, pden=1, outval=0, bank=2)
+ yield from gpios.config("4-7", oe=0, ie=1, puen=0, pden=1, outval=0, bank=6)
yield from gpios.set_out("0-3", outval=1)
#yield from gpios.rd_all()
projects / pinmux.git / blobdiff