Raspberry Pi Forums

Code: Select all

FUNCTIONS = {
    "I2C": {"enabled": False, "gpio": {0:"SDA", 1:"SCL", 2:"SDA", 3:"SCL"}, "modules": ["i2c-bcm2708", "i2c-dev"]},
    "SPI": {"enabled": False, "gpio": {7:"CE1", 8:"CE0", 9:"MISO", 10:"MOSI", 11:"SCLK"}, "modules": ["spi-bcm2708", "spidev"]},
    "UART": {"enabled": False, "gpio": {14:"TX", 15:"RX"}},
    "ONEWIRE": {"enabled": False, "gpio": {4:"DATA"}, "modules": ["w1-gpio"], "wait": 2}
}




def loadModule(module):
    debug("Loading module : %s" % module)
    subprocess.call(["modprobe", module])
    
def unloadModule(module):
    subprocess.call(["modprobe", "-r", module])
    
def loadModules(bus):
    if FUNCTIONS[bus]["enabled"] == False and not modulesLoaded(bus):
        info("Loading %s modules" % bus)
        for module in FUNCTIONS[bus]["modules"]:
            loadModule(module)
        if "wait" in FUNCTIONS[bus]:
            info("Sleeping %ds to let %s modules load" % (FUNCTIONS[bus]["wait"], bus))
            time.sleep(FUNCTIONS[bus]["wait"])

    FUNCTIONS[bus]["enabled"] = True

def unloadModules(bus):
    info("Unloading %s modules" % bus)
    for module in FUNCTIONS[bus]["modules"]:
        unloadModule(module)
    FUNCTIONS[bus]["enabled"] = False
        
def __modulesLoaded__(modules, lines):
    if len(modules) == 0:
        return True
    for line in lines:
        if modules[0].replace("-", "_") == line.split(" ")[0]:
            return __modulesLoaded__(modules[1:], lines)
    return False

def modulesLoaded(bus):
    if not bus in FUNCTIONS or not "modules" in FUNCTIONS[bus]:
        return True

    f = open("/proc/modules")
    c = f.read()
    f.close()
    lines = c.split("\n")
    return __modulesLoaded__(FUNCTIONS[bus]["modules"], lines)

def checkAllBus():
    for bus in FUNCTIONS:
        if modulesLoaded(bus):
            FUNCTIONS[bus]["enabled"] = True

class Bus():
    def __init__(self, busName, device, flag=os.O_RDWR):
        loadModules(busName)
        self.busName = busName
        self.device = device
        self.flag = flag
        self.fd = 0
        self.open()
        
    def __str__(self):
        return "Bus(%s, %s)" % (self.busName, self.device)
        
    def open(self):
        self.fd = os.open(self.device, self.flag)
        if self.fd < 0:
            raise Exception("Cannot open %s" % self.device)

    def close(self):
        if self.fd > 0:
            os.close(self.fd)
    
    def available(self):
        raise NotImplementedError
    
    def read(self, size=1):
        if self.fd > 0:
            return os.read(self.fd, size)
        raise Exception("Device %s not open" % self.device)
    
    def readBytes(self, size=1):
        return bytearray(self.read(size))
    
    def readByte(self):
        return self.readBytes()[0]

    def write(self, string):
        if self.fd > 0:
            return os.write(self.fd, string)
        raise Exception("Device %s not open" % self.device)
    
    def writeBytes(self, data):
        return self.write(bytearray(data))

    def writeByte(self, value):
        self.writeBytes([value])

Code: Select all

def request(method="GET", path="", data=None):
    def wrapper(func):
        func.routed = True
        func.method = method
        func.path = path
        func.data = data
        return func
    return wrapper

def response(fmt="%s", contentType="text/plain"):
    def wrapper(func):
        func.format = fmt
        func.contentType = contentType
        return func
    return wrapper


 @request("GET", "%(channel)d/value")
    @response("%d")
    def digitalRead(self, channel):
    self.checkDigitalChannel(channel)
    return self.__digitalRead__(channel)