rpirena/altera_ctrl.py

import struct
from time import sleep
import sys, os

import spidev
spidevname = "/dev/spidev0.0"
try:
    spi = os.open(spidevname, os.O_RDWR)
    spidev.set_mode(fd=spi, cpol=0, cpha=1)
except OSError as e:
    sys.stderr.write("%s: %s\n" % (spidevname, repr(e)))

def gpio_read(port):
    f = open("/sys/class/gpio/gpio%d/value" % port)
    r = int(f.readline())
    f.close()
    return r

def gpio_write(port, value):
    f = open("/sys/class/gpio/gpio%d/value" % port, "w")
    f.write("%d\n" % value)
    f.close()

nCONFIG=24
nSTATUS=25
CONF_DONE=23

def gpio_init(port, d=None, uwait=2):
    f = open("/sys/class/gpio/export", "w")
    f.write(str(port))
    try:
        f.close()
    except IOError as e:
        pass
    if d:
        try:
            f = open("/sys/class/gpio/gpio%d/direction" % port, "w")
        except IOError as e:
            # wait for assignment of ownership (by udev?)
            sleep(uwait)
            f = open("/sys/class/gpio/gpio%d/direction" % port, "w")
        f.write(d)
        f.close()

_verbosity=1
def set_verbosity(v, incr=False):
    global _verbosity
    if incr:
        _verbosity += v
    else:
        _verbosity = v

def altera_reset(timeout=10):
    gpio_init(nSTATUS)
    gpio_init(CONF_DONE)
    gpio_init(nCONFIG, "out")
    if _verbosity>=2:
        sys.stderr.write("nCONFIG=%d nSTATUS=%d CONF_DONE=%d\n" 
                         % (gpio_read(nCONFIG), 
                            gpio_read(nSTATUS), 
                            gpio_read(CONF_DONE),
                            ))
    gpio_write(nCONFIG, 0);
    nST = gpio_read(nSTATUS)
    n=0
    while nST and n<timeout:
        n += 1
        nST = gpio_read(nSTATUS)
    if _verbosity>=2:
        sys.stderr.write("nCONFIG=%d nSTATUS=%d CONF_DONE=%d\n" 
                         % (gpio_read(nCONFIG), 
                            nST, 
                            gpio_read(CONF_DONE),
                            ))
    gpio_write(nCONFIG, 1);
    if nST:
        raise IOError("nSTATUS does not go low")
    nST = gpio_read(nSTATUS)
    n=0
    while not nST and n<timeout:
        n += 1
        nST = gpio_read(nSTATUS)
    if _verbosity>=2:
        sys.stderr.write("nCONFIG=%d nSTATUS=%d CONF_DONE=%d\n" 
                         % (gpio_read(nCONFIG), 
                            nST, 
                            gpio_read(CONF_DONE),
                            ))
    if not nST:
        raise IOError("nSTATUS does not go high")

def altera_from_file(fn='rpirena.rbf', hz=30000000):
    altera_reset()
    spidev.set_mode(cpol=0, cpha=0)
    spidev.set_speed(hz=hz)
    f = open(fn, "rb")
    n = 0
    while True:
        b = f.read(1024)
        if not b:
            break
        n += len(b)
        spidev.sync_transfer(cmd=b, flags=spidev.INVERT_BITS_OUT, rsize=0)
        nST = gpio_read(nSTATUS)
        if not nST:
            IOError("nSTATUS went low during config")
        if _verbosity >= 2:
            CD = gpio_read(CONF_DONE)
            if _verbosity >= 3 or CD:
                sys.stderr.write("Altera config %d bytes, CONF_DONE=%d\n" 
                                 % (n, CD))
    sleep(1)
    CD = gpio_read(CONF_DONE)
    if not CD:
        raise IOError(f"CONF_DONE did not go high after config nSTATUS={gpio_read(nSTATUS)}")
    spidev.set_mode(cpol=0, cpha=1)

def cmdstr(w):
    if w & 0xffff4000:
        return struct.pack(">HH", w&0xffff, w>>16)
    return struct.pack(">H", w)

def sync(w):
    r = spidev.sync_transfer(cmd=cmdstr(w))
    if len(r)==4:
        return struct.unpack(">HH", r)
    return struct.unpack(">H", r)[0]

def reg(a, d=None):
    if d==None:
        c = struct.pack(">3H", a|0x8000, 0x8001, 0);
    else:
        c = struct.pack(">4H", a|0xc000, d, 0x8001, 0);
    return struct.unpack(">H", spidev.sync_transfer(cmd=c, rsize=2))[0]

def readfifo(fifo, magic=0, mask=0, size=0):
    if magic and not mask:
        mask = 0xffff
    n = reg(0x10 | fifo) & 0xfff
    c = struct.pack(f">{n+2}H", 0x8014 | fifo,  *((0x8001,)*n), 0)
    d = struct.unpack(f">{n}H", spidev.sync_transfer(cmd=c, rsize=2*n))
    i = 0
    while i < n and (d[i] & mask) != magic:
        i += 1
    if size and i + size > n:
        raise IOError(f"invalid packet({fifo}, {magic:04x}, {mask:04x}) → {repr(d)}")
    return d[i:]