python类I2C的实例源码

def __init__(self, type, scl, sda):
        self.type = type

        if self.type == 0:
            i2c_bus = I2C(scl=Pin(scl), sda=Pin(sda), freq=100000)
            self.sensor = BMP180.BMP180(i2c_bus)
            self.sensor.oversample_sett = 2
            self.sensor.baseline = 101325

        elif self.type == 1:
             pass #TODO

        else:
            log.error("Unknown sensor type '{}'. Cannot instantiate it.".format(self.type))

    # @timed_function

def run(owner):
    from utime import sleep
    sleep(0.1)
    from gc import collect
    from machine import I2C, Pin, RTC
    bus = I2C(-1, Pin(4), Pin(5))
    from urtc import DS3231
    rtc = RTC()
    dl = [x for x in DS3231(bus).datetime() if x]
    rtc.datetime(dl + [0 for x in range(0, 8 - len(dl))])
    del I2C, Pin, DS3231, rtc, dl
    collect()
    sleep(0.2)
    from gc import collect
    from states import init
    init_state = init()
    owner.core.load()
    collect()
    init_state.on_enter(owner)

def __init__(self,pin1='5',pin2='4'):
        """initialize the function with the pins 5,4 if you don't choice else
        fill pin,pin in when calling the function.
        In this function we initialize the i2c bus and the oled display. Than
        activate wifi radio. """
        self.pin1 = pin1
        self.pin2 = pin2
        self.name = ''
        self.strengt = ''
        self.status = ''
        self.kanaal = ''
        self.i2c = machine.I2C(machine.Pin(5), machine.Pin(4))
        self.oled = ssd1306.SSD1306_I2C(128, 64, self.i2c)
        self.oled.fill(1)
        self.oled.show()
        self.wlan = network.WLAN(network.STA_IF)
        self.wlan.active(True)

def main():
    while True:
        try:
            i2c = machine.I2C(scl=machine.Pin(0), sda=machine.Pin(2))
            bme = bme280.BME280(i2c=i2c, address=119)

            c.connect()
            c.publish(b"heartbeat", CLIENT_ID)

            c.publish(CLIENT_ID + "/temperature", bme.values[0])
            c.publish(CLIENT_ID + "/pressure", bme.values[1])
            c.publish(CLIENT_ID + "/humidity", bme.values[2])
            print("published msg")

        except Exception as e:
            print(str(e))
            machine.reset()
        finally:
            c.disconnect()
        time.sleep(5)

def __init__(self, address=0x20, gpioScl=5, gpioSda=4):
        """Initialize MCP230xx at specified I2C address and bus number.  If bus
        is not specified it will default to the appropriate platform detected bus.
        """
        self.address = address
        self.i2c = I2C(scl=Pin(gpioScl),sda=Pin(gpioSda))
        # Assume starting in ICON.BANK = 0 mode (sequential access).
        # Compute how many bytes are needed to store count of GPIO.
        self.gpio_bytes = self.NUM_GPIO//8
        # Buffer register values so they can be changed without reading.
        self.iodir = bytearray(self.gpio_bytes)  # Default direction to all inputs.
        self.gppu = bytearray(self.gpio_bytes)  # Default to pullups disabled.
        self.gpio = bytearray(self.gpio_bytes)
        # Write current direction and pullup buffer state.
        self.write_iodir()
        self.write_gppu()

def __init__(self, address=0x20, gpioScl=22, gpioSda=21):
        """Initialize MCP230xx at specified I2C address and bus number.  If bus
        is not specified it will default to the appropriate platform detected bus.
        """
        self.address = address
        self.i2c = I2C(scl=Pin(gpioScl),sda=Pin(gpioSda))
        # Assume starting in ICON.BANK = 0 mode (sequential access).
        # Compute how many bytes are needed to store count of GPIO.
        self.gpio_bytes = self.NUM_GPIO//8
        # Buffer register values so they can be changed without reading.
        self.iodir = bytearray(self.gpio_bytes)  # Default direction to all inputs.
        self.gppu = bytearray(self.gpio_bytes)  # Default to pullups disabled.
        self.gpio = bytearray(self.gpio_bytes)
        # Write current direction and pullup buffer state.
        self.write_iodir()
        self.write_gppu()

def send_cmd(self, cmd_request, response_size=6, read_delay_ms=100):
        """
        Send a command to the sensor and read (optionally) the response
        The responsed data is validated by CRC
        """
        try:
            self.i2c.start(); 
            self.i2c.writeto(self.i2c_addr, cmd_request); 
            if not response_size:
                self.i2c.stop();    
                return
            time.sleep_ms(read_delay_ms)
            data = self.i2c.readfrom(self.i2c_addr, response_size) 
            self.i2c.stop(); 
            for i in range(response_size//3):
                if not self._check_crc(data[i*3:(i+1)*3]): # pos 2 and 5 are CRC
                    raise SHT30Error(SHT30Error.CRC_ERROR)
            if data == bytearray(response_size):
                raise SHT30Error(SHT30Error.DATA_ERROR)
            return data
        except OSError as ex:
            if 'I2C' in ex.args[0]:
                raise SHT30Error(SHT30Error.BUS_ERROR)
            raise ex

def __init__(self, address=0x20, gpioScl=5, gpioSda=4):
        """Initialize MCP230xx at specified I2C address and bus number.  If bus
        is not specified it will default to the appropriate platform detected bus.
        """
        self.address = address
        self.i2c = I2C(scl=Pin(gpioScl),sda=Pin(gpioSda))
        # Assume starting in ICON.BANK = 0 mode (sequential access).
        # Compute how many bytes are needed to store count of GPIO.
        self.gpio_bytes = self.NUM_GPIO//8
        # Buffer register values so they can be changed without reading.
        self.iodir = bytearray(self.gpio_bytes)  # Default direction to all inputs.
        self.gppu = bytearray(self.gpio_bytes)  # Default to pullups disabled.
        self.gpio = bytearray(self.gpio_bytes)
        # Write current direction and pullup buffer state.
        self.write_iodir()
        self.write_gppu()

def bootstrap(self, timeout):
        from wifi import WiFi
        network = WiFi()
        from ssd1306 import SSD1306_I2C
        from machine import Pin, I2C
        bus = I2C(-1, Pin(4), Pin(5))
        display = SSD1306_I2C(128, 32, bus)
        from views import Views
        with network.Context(timeout) as c:
            while 1:
                ttl = self.ttl(timeout)
                if not ttl:
                    self._core.show(display, None)
                    self.data_changed = False
                    break
                self._core.show(
                    display,
                    Views['network']['wait'](c.net.token, timeout, ttl)
                )
                if c.net.connected:
                    self._core.show(
                        display,
                        Views['network']['connected']()
                    )
                    self.data_changed = self._core.setup_mode()
                    break
                self._sleep(0.5)
        del network, WiFi, display, SSD1306_I2C, Views, Pin, I2C, bus
        collect()
        return self.data_changed

def show_current_otp(self):
        start = self._get_time()
        from ssd1306 import SSD1306_I2C
        from machine import Pin, I2C
        bus = I2C(-1, Pin(4), Pin(5))
        display = SSD1306_I2C(128, 32, bus)
        from views import Views
        while self._get_time() - start <= OTP_SESSION:
            self._core.show(display, Views['otp'](
                self._core.get_otp_tuple()
            ))
            self._sleep(0.5)
        del display, SSD1306_I2C, Views, Pin, I2C, bus
        collect()

def turn_off(self):
        from ssd1306 import SSD1306_I2C
        from machine import I2C, Pin
        display = SSD1306_I2C(128, 32, I2C(-1, Pin(4), Pin(5)))
        display.poweroff()
        from machine import deepsleep
        if DEEPSLEEP:
            deepsleep()

def init_bmp(self):
        bus =  I2C(scl=Pin(self.BMPSCL), sda=Pin(self.BMPSDA), freq=100000)
        self.bmp = BMP180(bus)

def init_lcd(self):
        i2c = I2C(scl=Pin(self.DISSCL), sda=Pin(self.DISSDA), freq=400000)
        self.lcd = I2cLcd(i2c, self.DEFAULT_LCD_ADDR, 2, 16)

def main():
    i2c = machine.I2C(machine.Pin(5), machine.Pin(4))
    oled = ssd1306.SSD1306_I2C(128, 64, i2c)
    oled.fill(1)
    oled.show()
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    while True:
        try:
            wlan_list = wlan.scan()
        except:
            wlan_list = [['NONE','NONE','NONE','NONE','NONE','NONE']]
        for i in wlan_list:
            name = str(i[0], 'utf8')
            var = str(i[3])+' dBm'
            status = return_wifi_sec(i[4])
            kanaal = 'channel ' + str(i[2])
            oled.fill(0)
            oled.show()
            if len(name) > 15:
                oled.text(name[0:15],0,0)
                oled.text(name[15:int(len(name))],0,10)
            else:
                oled.text(name,0,0)
            oled.text(var,30,20)
            oled.text(status,30,30)
            oled.text(kanaal, 30,40)
            oled.show()
            utime.sleep_ms(10000)

def __init__(self, 
                 width = 128, height = 64, 
                 scl_pin_id = 5, sda_pin_id = 4,
                 freq = 400000): 

        self.width = width
        self.height = height
        self.i2c = machine.I2C(scl = machine.Pin(scl_pin_id, machine.Pin.OUT),
                               sda = machine.Pin(sda_pin_id), 
                               freq = freq) 
        self.display = ssd1306.SSD1306_I2C(width, height, self.i2c)
        self.show = self.display.show

def __init__(self, 
                 width = 128, height = 64, 
                 scl_pin_id = 5, sda_pin_id = 4,
                 freq = 400000): 

        self.width = width
        self.height = height
        self.i2c = machine.I2C(scl = machine.Pin(scl_pin_id, machine.Pin.OUT),
                               sda = machine.Pin(sda_pin_id), 
                               freq = freq) 
        self.display = ssd1306.SSD1306_I2C(width, height, self.i2c)
        self.show = self.display.show

def __init__(self, name, sda=None, scl=None,
                 width=128, height=64,
                 addr=0x3c):
        self.present = False
        self._y = 0
        self._x = 0
        self.last_text = ""
        self.char_width = width // 8
        self.char_height = height // 8

        if type(sda) is I2C:
            i2c = sda
        else:
            i2c = I2C(sda=sda, scl=scl)
        # test if lcd is responding
        try:
            self.dp = ssd1306.SSD1306_I2C(width, height, i2c, addr=addr)
            self.clear(show=False)
            self.println("  iot.ulno.net\n",show=True)

        except OSError:
            print("lcd not found")
        else:
            self.present = True
            Device.__init__(self, name, i2c, setters={"set":self.evaluate},
                            report_change=False)
            self.getters[""]=self.value

def __init__(self, name, sda=None, scl=None,
                 width=16, height=2,
                 addr=0x27):
        self.present = False
        self.last_text = ""
        self.char_width = width
        self.char_height = height
        self.display_buffer = bytearray(b" " * width * height)
        self.back_buffer = bytearray(b" " * width * height)
        self.x = 0
        self.y = 0

        if type(sda) is I2C:
            i2c = sda
        else:
            i2c = I2C(sda=sda, scl=scl)
        # test if lcd is responding
        try:
            self.dp = I2cLcd(i2c, addr, height, width)
            self.dp.clear()
            self.clear()
            self.hide_cursor()
            self.println("iot.ulno.net")

        except OSError:
            print("lcd not found")
        else:
            self.present = True
            Device.__init__(self, name, i2c, setters={"set":self.evaluate},
                            report_change=False)
            self.getters[""] = self.value

def __init__(self, name, sda=None, scl=None,
                 addr=8, ignore_case=False, report_change=True):
        not_found = "i2c device not found"

        self.addr = addr
        self.ownaddr = addr
        self.count = None
        self.current_value = ""
        self.suspend_start = None
        self.suspend_time = 0
        self.msgq = None
        if type(sda) is I2C:
            i2c = sda
        else:
            i2c = I2C(sda=sda, scl=scl)

        try:
            l = i2c.scan()  # see, if you can find the dive with respective addr
        except OSError:
            print(not_found)
        else:
            if addr in l:
                self.present = True
                Device.__init__(self, name, i2c, setters={"set":self.evaluate},
                                ignore_case=ignore_case,report_change=report_change)
                self.getters[""]=self.value
            else:
                print(not_found)

def __init__(self, i2c_sda_pin, i2c_scl_pin):
        self.i2c = I2C(sda=Pin(i2c_sda_pin), scl=Pin(i2c_scl_pin))
        self.addr = 90 # I2C address of the MPR121
        # enable ELE0 - ELE3
        self.enable_elec(ELEC_ENABLE_DEFAULT)

def __init__(self, scl_pinno=5, sda_pinno=4):
        self.i2c = I2C(scl=Pin(scl_pinno, Pin.IN),
                       sda=Pin(sda_pinno, Pin.IN))

def __init__(self, sensor_id='mcp9808',
                 address=MCP9808_I2CADDR_DEFAULT, scl_pinno=5, sda_pinno=4,
                 i2c=None):
        """Initialize MCP9808 device on the specified I2C address and bus number.
        Address defaults to 0x18 and bus number defaults to the appropriate bus
        for the hardware.
        """
        self.sensor_id = sensor_id
        if i2c is None:
            self.i2c = I2C(scl=Pin(scl_pinno, Pin.IN),sda=Pin(sda_pinno, Pin.IN))
        else:
            self.i2c = i2c
        self.address=address
        assert self.begin(), "Invalid values read from I2C bus for MCP9808"

def __init__(self, scl_pinno=5, sda_pinno=4):
        self.i2c = I2C(scl=Pin(scl_pinno, Pin.IN),
                       sda=Pin(sda_pinno, Pin.IN))

def writeList(self, register, data):
        """Introduced to match the writeList implementation of the Adafruit I2C _device member"""
        return self.i2c.writeto_mem(self.address, register, data)

def readList(self, register, length):
        """Introduced to match the readList implementation of the Adafruit I2C _device member"""
        return self.i2c.readfrom_mem(self.address, register, length)

def __init__(self, gpio_interface, scl_pin=5, sda_pin=4, frequency=100000):

        gpio_interface.setup_pin(scl_pin)
        gpio_interface.setup_pin(sda_pin)

        self._scl = gpio_interface.get_pin(scl_pin)
        self._sda = gpio_interface.get_pin(sda_pin)

        self._bus = I2C(scl=self._scl, sda=self._sda, freq=frequency)

def __init__(self):
        self.i2c =I2C(scl=Pin(14),sda=Pin(2),freq=100000)

def writeList(self, register, data):
        """Introduced to match the writeList implementation of the Adafruit I2C _device member"""
        return self.i2c.writeto_mem(self.address, register, data)

def readList(self, register, length):
        """Introduced to match the readList implementation of the Adafruit I2C _device member"""
        return self.i2c.readfrom_mem(self.address, register, length)

def setup():
    global i2c
    global oled
    i2c = I2C(scl=Pin(22), sda=Pin(21), freq=100000)
    oled = ssd1306.SSD1306_I2C(128, 64, i2c,  addr=0x3c, external_vcc=False)