Python machine 模块，Pin() 实例源码

def write(self, payload):

    new_state = None

    if payload in ['1', 'true', 'on']:
      new_state = True
    elif payload in ['0', 'false', 'off' ]:
      new_state = False
    elif payload == "toggle":
      new_state = not bool(self.pin.value())
    else: 
      return { "error": "Unsupported state: {}".format(payload) }

    self.pin.value(new_state)

#    Util.log(self,"write final state '{}' payload: '{}'  pin 15 state: '{}'".format(self.pin.value(), payload,machine.Pin(15, machine.Pin.OUT).value()))

    return self.read()

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 _config(self, duty_cycle=0):
        if self._function == 'dig':
            _mode = machine.Pin.OUT if self._mode == 'out' else machine.Pin.IN
            if self._pull == 'pu':
                _pull = machine.Pin.PULL_UP
            elif self._pull == 'pd':
                _pull = machine.Pin.PULL_DOWN
            else:
                _pull = None
            self._pin = machine.Pin(self._name, mode=_mode, pull=_pull, drive=machine.Pin.MED_POWER)
        elif self._function == 'ana':
            adc = machine.ADC(bits=12)
            self._apin = adc.channel(pin=self._name)
        else:
            self._pwm = machine.PWM(machine.Pin(self.pin_num),duty=duty_cycle * 100,freq=1000)
            #timer = machine.Timer(HwPin._TimerMap[self._name][0], mode=machine.Timer.PWM)
            #self._pwm = timer.channel(HwPin._TimerMap[self._name][1], freq=20000, duty_cycle=(duty_cycle * 100))

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 read_temp():
    # the device is on GPIO12
    dat = machine.Pin(5)

    # create the onewire object
    ds = ds18x20.DS18X20(onewire.OneWire(dat))

    # scan for devices on the bus
    roms = ds.scan()
    # print('found devices:', roms)
    # print('temperature:', end=' ')
    ds.convert_temp()
    time.sleep_ms(750)
    temp = ds.read_temp(roms[0])
    # print(temp)
    return temp

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 readDS18x20():
    from machine import Pin
    import onewire, ds18x20

    OneWirePin = 0

    # the device is on GPIO12
    dat = Pin(OneWirePin)
    # create the onewire object
    ds = ds18x20.DS18X20(onewire.OneWire(dat))
    # scan for devices on the bus
    roms = ds.scan()
    ds.convert_temp()
    time.sleep_ms(750)
    values = []

    for rom in roms:
        values.append(ds.read_temp(rom))

    print(values)
    return values

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 __init__(self, gc_enable=True, heartbeat=None):
        self.lstThread = []                     # Entries contain [Waitfor object, function, pid, state, due]
        self.add_thread_bar = False             # Re-entrancy check
        self.bStop = False
        self.last_gc = 0
        self.pid = 0
        self.gc_enable = gc_enable
        self.last_heartbeat = 0
        self.heartbeat = heartbeat
        if heartbeat is not None:
            if platform == 'pyboard':
                if heartbeat > 0 and heartbeat < 5:
                    import pyb
                    self.heartbeat = pyb.LED(heartbeat)
                else:
                    raise ValueError('heartbeat must be a valid LED no.')
            elif platform == 'esp8266':
                import machine
                self.heartbeat = machine.Pin(2, machine.Pin.OUT)

def tick(self, pin=2) :
        import dht
        import machine
        try :
            d = dht.DHT11(machine.Pin(pin))
            d.measure()
            tempf = 32.0 + 1.8 * d.temperature()
            humidity = d.humidity()
            logging.debug("Read measurements off DHT11: temp(f): %s humidity: %s" % (tempf, humidity))
            self._upload(tempf, humidity)
            util.clear_led_error()
        except Exception as E :
            logging.error("An error occurred taking measurements: %s", E)
            util.set_led_error()

    ##
    ## Internal operations
    ##

def status(self):
        ow = onewire.OneWire(Pin(2))
        ds = ds18x20.DS18X20(ow)
        roms = ds.scan()

        if not len(roms):
            print("Nao encontrei um dispositivo onewire.")
            return None

        ds.convert_temp()
        time.sleep_ms(750)

        temp = 0
        for i in range(10):
            for rom in roms:
                temp += ds.read_temp(rom)
        return temp / 10

    # envia a temperatura para o broker

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 configure(self, config):
    super().configure(config)
    self.pin = machine.Pin(self.config["gpio"], machine.Pin.OUT)

    if self.config['state'] is not None:
      self.pin.value(bool(self.config['state']))

def configure(self, config):
    super().configure(config)
    self.pin = machine.Pin(self.config["gpio"], machine.Pin.OUT)
#    self.pin = machine.Pin(self.config["gpio"], machine.Pin.OPEN_DRAIN)

def configure(self, config):
    super().configure(config)
    self.pin = machine.Pin(self.config["gpio"], machine.Pin.OUT)

def configure(self, config):
    super().configure(config)
    self.pin = machine.Pin(self.config["gpio"], machine.Pin.OPEN_DRAIN)

    if self.config['state'] is not None:
      self.pin.value(bool(self.config['state']))

def __init__(self, type, pin):
        self.type = type

        if self.type == 0:
            self.sensor = dht.DHT11(Pin(pin))
        elif self.type == 1:
             self.sensor = dht.DHT22(Pin(pin))
        else:
            log.error("Unknown sensor type '{}'. Cannot instantiate it.".format(self.type))

    # @timed_function

def toggle_door():
    p = machine.Pin(DOOR_TOGGLE_PIN, machine.Pin.OUT)
    p.value(1)
    await asyncio.sleep_ms(100)
    p.value(0)


# open_door and close_door just call toggle_door for now, since my
# garage door opener uses the same mechanism to do both. keeping these separate
# for, uh, modularity.

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  ledR():
    log.info("Starting ledR")
    led    = machine.Pin(5, mode=machine.Pin.OUT)  
    total = 0
    yield
    while True:
        total += 1
        if led.value() == 1:
            led.value(0)
        else:
            led.value(1)    
        yield

# 2 tasks

def __init__(self, controller="XPT2046", asyn=False, *, confidence=5, margin=50, delay=10, calibration=None, spi = None):
        if spi is None:
            self.spi = SPI(-1, baudrate=1000000, sck=Pin("X12"), mosi=Pin("X11"), miso=Pin("Y2"))
        else:
            self.spi = spi
        self.recv = bytearray(3)
        self.xmit = bytearray(3)
# set default values
        self.ready = False
        self.touched = False
        self.x = 0
        self.y = 0
        self.buf_length = 0
        cal = TOUCH.DEFAULT_CAL if calibration is None else calibration
        self.asynchronous = False
        self.touch_parameter(confidence, margin, delay, cal)
        if asyn:
            self.asynchronous = True
            import uasyncio as asyncio
            loop = asyncio.get_event_loop()
            loop.create_task(self._main_thread())

# set parameters for get_touch()
# res: Resolution in bits of the returned values, default = 10
# confidence: confidence level - number of consecutive touches with a margin smaller than the given level
#       which the function will sample until it accepts it as a valid touch
# margin: Difference from mean centre at which touches are considered at the same position
# delay: Delay between samples in ms.
#

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

def init_dht(self):
        self.dht = dht.DHT22(Pin(self.DHTPIN))

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 _add_to_pin_mask(self, mask, pin):
        if pin == 'P10' or pin == 'G17':
            mask |= 0x01
        elif pin == 'P17' or pin == 'G31':
            mask |= 0x02
        elif pin == 'P18' or pin == 'G30':
            mask |= 0x08
        else:
            raise ValueError('Invalid Pin specified: {}'.format(pin))
        return mask

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 main():
    p = machine.Pin(2, machine.Pin.OUT)
    p.value(0)
    loop.create_task(reconnect(umqtt_lock))
    loop.create_task(check_subs(umqtt_lock))
    loop.create_task(publish_status(umqtt_lock))
    loop.run_forever()

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 read_temps():
    """Read DS18B20's."""
    dat = machine.Pin(PIN_1W)
    ds = ds18x20.DS18X20(onewire.OneWire(dat))

    ds.convert_temp()
    time.sleep_ms(750)

    return {rom_to_hex(rom): ds.read_temp(rom) for rom in ds.scan()}

def f(t):
    d=dht.DHT11(machine.Pin(4))
    d.measure()
    a=d.temperature()
    b=d.humidity()
    print('¶:',a,'C')
    print('?:',b,'%')

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

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 setup():
    global d
    d = dht.DHT22(Pin(2)) 
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    cfg = load_config()
    ap = cfg['ap']
    wlan.connect(ap['ssid'], ap['pwd'])
    while not wlan.isconnected():
        machine.idle()
    wifi(cfg['mqtt'])

def set_led_error(pin_id=15) :
    import machine
    pin = machine.Pin(pin_id, machine.Pin.OUT)
    pin.high()

def clear_led_error(pin_id=2) :
    import machine
    pin = machine.Pin(pin_id, machine.Pin.OUT)
    pin.low()

def dht_test(pin=2, n=1, sleep_s=5) :
    import dht
    import machine
    d = dht.DHT11(machine.Pin(pin))
    for i in range(n):
        d.measure()
        line = "temp(f): %s humidity: %s" % (32.0 + 1.8*d.temperature(), d.humidity()) 
        print(line)
        #upload("192.168.1.154", 44404, line)
        upload("KMABOXBO9", "7rwzw8bh", 32.0 + 1.8*d.temperature(), d.humidity())
        #print("Going into deep sleep for %s secs ..." % sleep_secs)
        #deep_sleep(sleep_secs)
        print("Sleeping %s secs ..." % sleep_s)
        time.sleep(sleep_s)

def blink(pin_id) :
    import machine
    pin = machine.Pin(pin_id, machine.Pin.OUT)
    while True :
        pin.high()
        time.sleep(1)
        pin.low()
        time.sleep(1)

def clear_led_error(pin_id=14) :
    import machine
    pin = machine.Pin(pin_id, machine.Pin.OUT)
    pin.low()

def create_neopixel(pin, num_pixels) :
        return neopixel.NeoPixel(machine.Pin(pin), num_pixels)

def set_spi_irq_as_output(self):
        """Pull IRQ high"""
        self._irq_pin.init(mode=machine.Pin.OUT_PP,
                           pull=machine.Pin.PULL_NONE, value=1)

def set_spi_irq_as_input(self):
        """IRQ input"""
        self._irq_pin.init(mode=machine.Pin.IN, pull=machine.Pin.PULL_DOWN)

def __init__(self, pin=12, place='', server='localhost', chipid='', mac=''):
      self.count = 0
      self.sensor = '000'
      self.temp = '85.0'    # the default error temperature of ds18b20
      self.place = place
      self.server = server
      self.chipid = chipid
      self.mac = mac
      try:
          ow = OneWire(Pin(pin))
          self.ds = ds18x20.DS18X20(ow)
          self.roms = self.ds.scan()
          self.present = True
      except:
          self.present = False

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=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 )