Python machine 模块，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)

def init_i2c(self):
        print('* initializing i2c')
        self.bus = I2C(scl=self.pin_scl,
                       sda=self.pin_sda)

def __init__(self):
        self.d = None
        try:
            import ssd1306
            from machine import I2C, Pin
            i2c = I2C(sda=Pin(4), scl=Pin(5))
            self.d = ssd1306.SSD1306_I2C(64, 48, i2c, 60)
            print ("Display available")
        except Exception as e:
            print ("Display just print")
            self.d = None

def __init__(self, i2c, address=0x20 ):
        """Initialize MCP230xx at specified I2C address on the I2C Bus."""
        self.address = address
        self.i2c = i2c
        # 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 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, i2c=None, addr=AM2315_I2CADDR ):
        self.addr = addr
        self.rbuf = bytearray(8) # sensor response
        if i2c != None:
            self.i2c = i2c
        else:
            # WARNING this default bus does not work 
            # on feather ESP8266 with external power supply 
            # see test.py using sda on pin 4, scl on pin 2
            self.i2c = I2C( sda=Pin(4), scl=Pin(5), freq=20000 )

def __init__(self, scl_pin=5, sda_pin=4, delta_temp = 0, delta_hum = 0, i2c_address=DEFAULT_I2C_ADDRESS):
        self.i2c = I2C(scl=Pin(scl_pin), sda=Pin(sda_pin))
        self.i2c_addr = i2c_address
        self.set_delta(delta_temp, delta_hum)
        time.sleep_ms(50)

def init(self, scl_pin=5, sda_pin=4):
        """
        Init the I2C bus using the new pin values
        """
        self.i2c.init(scl=Pin(scl_pin), sda=Pin(sda_pin))

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, connect=None, *, pwr=None, i2c=None, spi=None, i2c_addr=98):
        if connect in ('X', 'Y', 'XY', 'YX'):
            i = connect[-1]
            j = connect[0]
            y = j + '4'
        elif connect == 'C':
            i = 2
            j = 2
            y = 'A7'
        else:
            if pwr is None or i2c is None or spi is None:
                raise ValueError('must specify valid "connect" or all of "pwr", "i2c" and "spi"')

        if pwr is None:
            pwr = machine.Pin(y, machine.Pin.OUT)
        if i2c is None:
            i2c = machine.I2C(i, freq=1000000)
        if spi is None:
            spi = machine.SPI(j, baudrate=13500000, polarity=0, phase=0)

        if not pwr.value():
            pwr(1)
            sleep_ms(10)
        # else:
        # alread have power
        # lets be optimistic...

        # set connections
        self.pwr = pwr
        self.i2c = i2c
        self.spi = spi
        self.i2c_addr = i2c_addr

        # create temp buffers and memoryviews
        self.buf16 = bytearray(16)
        self.buf19 = bytearray(19)
        self.buf = [None] * 10
        for i in range(1, 10):
            self.buf[i] = memoryview(self.buf16)[0:i]
        self.buf1 = self.buf[1]
        self.array4 = [0, 0, 0, 0]

        # set default orientation and window
        self.set_orient(PORTRAIT)
        self._fcmd2b('<BBBBBB', 0x76, 0, 0, self.w, self.h) # viewport 'v'
        self._fcmd2b('<BBBBBB', 0x79, 0, 0, self.w, self.h) # window 'y'