django-relay-controller/relaycontroller/modules/gpiolib.py

'''
Created on 21 août 2022

@author: denis
'''
import logging
import os
import threading


logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

if os.getenv('BOARD_DEVICE') == 'RPI':
    print("found a RPI board")
    import RPi.GPIO as GPIO
    logger.info(GPIO.RPI_INFO)  # @UndefinedVariable
    GPIO.setmode(GPIO.BOARD)    # @UndefinedVariable

elif os.getenv('BOARD_DEVICE') == 'RPI1B':
    print("found a RPI 1B board")
    import RPi.GPIO as GPIO     # @Reimport
    logger.info(GPIO.RPI_INFO)  # @UndefinedVariable
    GPIO.setmode(GPIO.BOARD)    # @UndefinedVariable

elif os.getenv('BOARD_DEVICE') == 'OPIWIN+':
    print("found an OPI WIN+ board")
    import orangepi.winplus
    from OPi import GPIO     # @Reimport
    GPIO.setmode(orangepi.winplus.BOARD)

elif os.getenv('BOARD_DEVICE') == 'OPIZERO':
    print("found an OPI ZERO board")
    import orangepi.zero
    from OPi import GPIO     # @Reimport
    GPIO.setmode(orangepi.zero.BOARD)

GPIO.setwarnings(False)      # @UndefinedVariable

IO_RISING = GPIO.RISING     # @UndefinedVariable
IO_FALLING = GPIO.FALLING   # @UndefinedVariable
IO_BOTH = GPIO.BOTH         # @UndefinedVariable
IO_LOW  = GPIO.LOW          # @UndefinedVariable
IO_HIGH = GPIO.HIGH         # @UndefinedVariable
IO_INPUT  = GPIO.IN         # @UndefinedVariable
IO_OUTPUT = GPIO.OUT        # @UndefinedVariable

IO_PULL_UP = GPIO.PUD_UP
IO_PULL_DOWN = GPIO.PUD_DOWN
IO_PULL_OFF = GPIO.PUD_OFF

IO_CLEANUP = GPIO.cleanup   # @UndefinedVariable

lock = threading.Lock()


def bitread(b, bitpos):
    return (b>>bitpos) & 0x1

class Pin:
    def __init__(self, channel, sens, initial=None, callback=None, edge=IO_FALLING, bouncetime=500, pull_up_down=GPIO.PUD_OFF):
        self.channel = channel
        self.sens = sens

        lock.acquire()
        if initial in [GPIO.LOW, GPIO.HIGH] and sens==GPIO.OUT:
            GPIO.setup(channel, sens, initial=initial, pull_up_down=pull_up_down)
        else:
            GPIO.setup(channel, sens, pull_up_down=pull_up_down)

        if callback is not None:
            self.set_interrupt(edge, callback, bouncetime)
        lock.release()

        logger.info(f'Channel {self.channel} sens: {sens}  initial: {initial}')
        self.value = self.read() # initial value


    def cleanup(self):
        lock.acquire()
        GPIO.cleanup(self.channel)
        lock.release()


    def set_interrupt(self, edge, callback, bouncetime=None):
        if callable(callback) and edge in [IO_RISING, IO_FALLING, IO_BOTH]:
            GPIO.add_event_detect(self.channel, edge, callback=callback, bouncetime=bouncetime)


    def remove_interrupt(self):
        lock.acquire()
        GPIO.remove_event_detect(self.channel)
        lock.release()


    def state(self):
        return self.value

    def read(self):
        lock.acquire()
        self.value = GPIO.input(self.channel)
        lock.release()
        #logger.info(f'Read {self.channel} -> {self.value}')
        return self.value

    def write(self, value):
        lock.acquire()
        GPIO.output(self.channel, value)
        lock.release()
        #logger.info(f'Write {self.channel} <- {value}')
        return self.read()


class ShiftInRegister74HC165():
    lastState = 0;
    currentState = 0;

    def __init__(self, latch_pin, clock_pin, data_pin, number=1):
        self.datalen   = number * 8
        self.latch_pin = Pin(latch_pin, IO_OUTPUT, initial=IO_LOW, pull_up_down=IO_PULL_DOWN)
        self.clock_pin = Pin(clock_pin, IO_OUTPUT, initial=IO_LOW, pull_up_down=IO_PULL_DOWN)
        self.data_pin  = Pin(data_pin, IO_INPUT, initial=IO_LOW, pull_up_down=IO_PULL_DOWN)
        threading.Event().wait(0.5)

    def read(self):
        self.lastState = self.currentState
        result = 0
        self.latch_pin.write(IO_HIGH)
        for i in range(self.datalen):
            value = self.data_pin.read()
            result |= (value << ((self.datalen-1) - i))
            self.clock_pin.write(IO_HIGH)
            self.clock_pin.write(IO_LOW)
        self.latch_pin.write(IO_LOW)
        self.currentState = result
        return result

    def state(self, idx):
        return bitread(self.currentState, idx)

    def last(self, idx):
        return bitread(self.lastState, idx)

    def update(self):
        return self.read()!=self.lastState

    def hasChanged(self, idx=None):
        return self.lastState!=self.currentState if idx is None else self.state(idx)!=self.last(idx)

    def pressed(self, idx):
        return not self.last(idx) and self.state(idx)

    def released(self, idx):
        return self.last(idx) and not self.state(idx)

    def cleanup(self):
        self.latch_pin.cleanup()
        self.clock_pin.cleanup()
        self.data_pin.cleanup()


class ShiftOutRegister74HC595():

    def __init__(self, latch_pin, clock_pin, data_pin, number=1):
        self.datalen   = number * 8
        self.latch_pin = Pin(latch_pin, IO_OUTPUT, initial=IO_LOW)
        self.clock_pin = Pin(clock_pin, IO_OUTPUT, initial=IO_LOW)
        self.data_pin  = Pin(data_pin, IO_OUTPUT, initial=IO_LOW)
        self.set_low()
        threading.Event().wait(0.5)

    def set_low(self):
        self.registers = [0] * self.datalen
        self.write()

    def write(self):
        self.clock_pin.write(IO_LOW)
        self.latch_pin.write(IO_LOW)
        self.clock_pin.write(IO_HIGH)
        #
        #for i in range(self.datalen):
        for bit in self.registers:
            self.clock_pin.write(IO_LOW)
            self.data_pin.write(int(bit))
            self.clock_pin.write(IO_HIGH)
        #
        self.clock_pin.write(IO_LOW)
        self.latch_pin.write(IO_HIGH)
        self.clock_pin.write(IO_HIGH)

    def set(self, index, state):
        self.registers[index] = int(state)
        self.write()
        return int(state)

    def get(self, index):
        return self.registers[index]

    def cleanup(self):
        self.set_low()
        self.latch_pin.cleanup()
        self.clock_pin.cleanup()
        self.data_pin.cleanup()