树莓派-Raspberry Pi Pico RP2040 使用大全（硬件接口功能、传感器模块等）_树莓派rp2040

主板

主板实物图

https://pico.org.cn/

主板引脚图

Thonny开发IDE工具

https://thonny.org

配置解释器：Raspberry Pi Pico
工具=》选项=》解释器（选择Raspberry Pi Pico）

查看帮助信息

help(‘modules’) ：查看可以通过import导入的模名

help(模块名)：查看指定模块下的类、函数、常量：

help(模块名.类)：查看指定模块类包含的函数和常量

Micropython官网文档

micropython参考文档

CPU主频

• RP2040主频最高支持133MHz

import machine

print(machine.freq()) # 打印当前CPU频率
machine.freq(24000000) # 设置CPU频率
1
2
3
4

GPIO 输出

GPIO输出模式：machine.Pin.OUT
GPIO输入模式：machine.Pin.IN

import machine
import utime

led = machine.Pin(25, machine.Pin.OUT)
print('gpio output demo...')
while True:
    led.value(0)
    utime.sleep(1)
    led.value(1)
    utime.sleep(0.5)
    
	# 其它GPIO相关操作接口
	#led.low()
    #led.high()
    #led.toggle()
    #led.on()
    #led.off()
    #led.value(0)
    #led.value(1)
    #state=led.value()
    #print(state)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

GPIO 输入

上拉：machine.Pin.PULL_UP
下拉：machine.Pin.PULL_DOWN

import machine
import utime

key = machine.Pin(23, machine.Pin.IN, machine.Pin.PULL_UP)
led = machine.Pin(25, machine.Pin.OUT)

print('gpio input demo...')
while True:
    if key.value() == 1:
        led.value(1)
    else:
        led.value(0)
    utime.sleep_ms(100)
        
1
2
3
4
5
6
7
8
9
10
11
12
13
14

GPIO 中断

irq(trigger, handler), trigger中断触发方式，handler中断处理函数
下降沿：trigger=machine.Pin.IRQ_FALLING
上升沿：trigger=machine.Pin.IRQ_RISING

import machine
import utime

key = machine.Pin(23, machine.Pin.IN, machine.Pin.PULL_DOWN)
led = machine.Pin(25, machine.Pin.OUT)

print('gpio irq demo...')
def irq_handler(pin):
    print(pin)
    led.toggle()

key.irq(trigger=machine.Pin.IRQ_FALLING, handler=irq_handler)

while True:
    utime.sleep(1)
    print("main loop...")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

ADC（模数转换）

外部ADC模拟电压表

import machine
import utime
import math

adc0 = machine.ADC(26)
adc1 = machine.ADC(27)
adc2 = machine.ADC(28)

adcvalue0 = 0
adcvalue1 = 0
adcvalue2 = 0
factor = 3.3/(65536)

while True:
    adcvalue0 = adc0.read_u16()*factor
    adcvalue1 = adc1.read_u16()*factor
    adcvalue2 = adc2.read_u16()*factor
    print(adcvalue0, adcvalue1, adcvalue2)
    utime.sleep_ms(500)    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

内部ADC测量温度

import machine
import utime

sensor_temp = machine.ADC(4)
#sensor_temp = machine.ADC(machine.ADC.CORE_TEMP)
conversion_factor = 3.3/(65535)
while True:
    reading = sensor_temp.read_u16()*conversion_factor
    temperature = 27 - (reading - 0.706)/0.001721
    print(temperature,"C")
    utime.sleep(1)
1
2
3
4
5
6
7
8
9
10
11

PWM（脉宽调制）

PWM相关接口：
pwm0 = machine.PWM(machine.Pin(0))
pwm0.freq() # 返回当前PWM设置频率
pwm0.freq(2000) # 设置PWM频率
pwm0.duty_u16(15000) # 设置占空比（0~65535）
pwm0.duty_ns(10) # 设置占空比时间ns

import machine
import utime
import urandom

led_r = machine.PWM(machine.Pin(4))
led_g = machine.PWM(machine.Pin(3))
led_b = machine.PWM(machine.Pin(2))

led_r.freq(1000)
led_g.freq(1000)
led_b.freq(1000)

print(led_r.freq())
print(led_g.freq())
print(led_b.freq())

while True:
    led_r.duty_u16(urandom.randrange(1, 65535, 10))
    led_g.duty_u16(urandom.randrange(1, 65535, 20))
    led_b.duty_u16(urandom.randrange(1, 65535, 30))    
    utime.sleep_ms(100)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

WDT看门狗

import machine
import utime

# enable the WDT with a timeout of 5s (1s is the minimum)
wdt = machine.WDT(timeout=5000)

print("WDT Demo...")
while True:
    wdt.feed()
    utime.sleep(3)
1
2
3
4
5
6
7
8
9
10

RTC实时时钟

rtc = machine.RTC()
rtc.datetime()返回一个元组数据：
(年，月，日，星期，时，分，秒，0)
(2022, 10, 14, 4, 14, 41, 57, 0)

import machine
import utime

rtc = machine.RTC()

# 设置RTC日期时间等参数
# rtc.datetime((2020, 1, 21, 2, 10, 32, 36, 0))

while True:
    print(rtc.datetime())
    utime.sleep(1)
1
2
3
4
5
6
7
8
9
10
11

I2C总线接口

I2C总线的通讯频率:

• 标准模式(Standard)：100kbps
• 快速模式(Fast)：400kbps
• 快速模式+(Fast-Plus)：1Mbps
• 高速模式(High-speed)：3.4Mbps
• 超快模式(Ultra-Fast)：5Mbps（单向传输）

构造函数

• machine.I2C(id, *, scl, sda, freq=400000)
  使用以下参数构造并返回一个新的 I2C 对象：
  id:标识特定的 I2C 外设。允许的值取决于特定的端口/板
  scl:应该是一个 pin 对象，指定用于 SCL 的 pin。
  sda:应该是一个 pin 对象，指定用于 SDA 的 pin。
  freq:应该是一个整数，用于设置 SCL 的最大频率。

一般方法

• i2c.scan(): 扫描所有 0x08 和 0x77 之间的 I2C 地址，然后返回一个有响应地址的列表

标准总线操作

• i2c.readfrom(addr, nbytes, stop=True):

  • 功能：从addr指定的从站读取nbytes
  • addr：从设备地址
  • nbytes：指定读取的字节数
  • stop：True传输结束时发送停止信号
  • 返回读取数据的对象
  • buf = i2c.readfrom(0x44, 5)

• i2c.readfrom_into(addr, buf, stop=True):

  • 功能：从addr指定的从设备读入buf长度的数据到buf内
  • addr：从设备地址
  • stop：True传输结束时发送停止信号
  • 返回: None
  • buf = bytearray(3)
    i2c.readfrom_into(0x44, buf)

• i2c.writeto(addr, buf, stop=True)：

  • 功能：将buf 中的字节写入addr指定的从设备。
  • addr：从设备I2C地址
  • buf：要写入的字节数组，存放写入的数据，写入的长度为buf的长度。
  • stop：True传输结束时发送停止信号
  • 返回接收到的ACK数
  • 示例：
    buf=bytearray([0x01,0x02,0x03])
    i2c.writeto(0x74, buf) # 向从设备地址为0x74设备写入3个字节数据

• i2c.writevto(addr, vector，stop:True)：新增更高效的大量数据写入操作。（设备地址 + 命令 + 数据）

  • 功能：将vector 中包含的字节写入addr指定的从站
  • addr：设备地址
  • vector：vector应该是具有缓冲协议的元组或对象列表
  • stop：True传输结束时发送停止信号
  • 返回接收到的 ACK 数
  • 示例：
    write_list = [b"\x40", None] # b"\x40" : 命名
    write_list[1] = bytearray([0x01,0x05,0x06,0x08]) # 数据
    ret = i2c.writevto(0x44, write_list)
    print(ret) # 写入字节数 5 个

内存操作

• i2c.readfrom_mem(addr, memaddr, nbytes, *, addrsize=8):

  • 功能：从addr指定的从设备的memaddr指定的内存地址开始，读取nbytes。
  • addr：从设备地址
  • memaddr：寄存器地址
  • nbytes：读取字节数
  • addrsize：寄存器地址长度
  • 返回读取数据的对象
  • buf = i2c.readfrom_mem(0x44, 0x3244, 5, addrsize=16)

• i2c.readfrom_mem_into(addr, memaddr, buf, *, addrsize=8):

  • 功能：从addr指定的从设备的memaddr内存地址开始读入buf长度的数据到buf。
  • addr：从设备地址
  • memaddr：寄存器地址
  • buf：存放读取数据的字节数组
  • addrsize：寄存器长度
  • 返回None
  • buf = bytearray(10)
    i2c.readfrom_mem_into(0x44, 0x33, buf, addrsize=8)

• i2c.writeto_mem(addr, memaddr, buf, *, addrsize=8)：

  • 功能：将buf中的数据写入到addr从设备的 memaddr开始的寄存器地址中。
  • addr：从设备地址
  • memaddr：寄存器地址
  • buf：写入的数据字节数组
  • addrsize：寄存器地址长度，8：单字节，16：双字节
  • 示例：
    buf = bytearray([0x01,0x02,0x03])
    i2c.writeto_mem(0x44, 0xCC66, buf, addrsize=16)

最原始的通信方式：

 def write_data(self, buf):
     self.temp[0] = self.addr << 1
     self.temp[1] = 0x40
     self.i2c.start()       
     self.i2c.write(self.temp)       
     self.i2c.write(buf)       
     self.i2c.stop()
1
2
3
4
5
6
7

搜索I2C总线设备：

from machine import Pin, I2C
import utime

i2c = I2C(0, sda=Pin(0), scl=Pin(1), freq=400000)
devlist = i2c.scan()
print(devlist)
for dev in devlist:
    print(hex(dev))
1
2
3
4
5
6
7
8

读写I2C设备示例：

import machine
import utime

sda = machine.Pin(16)
scl = machine.Pin(17)
i2c = machine.I2C(0, sda=sda, scl=scl, freq=200000)
  
print(i2c.scan())

i2c.writeto(35, '0x01') # power on
i2c.writeto(35, '0x10') # H-resolution mode
i2c.writeto(35, '0x20')
utime.sleep(0.2)



while True:
    i2c.writeto(35, '\x01')
    val = i2c.readfrom(35, 2, True)
    print(val)
    utime.sleep(1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

UART串口通信

串口通信相关接口：
rxbuf=bytes()
rxData=bytes()
txData=bytes()
tx = machine.Pin(16)
rx = machine.Pin(17)
uart = machine.UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=tx, rx=rx)
uart.any() # True:有可以读取的数据，False：没有可读取的数据
rxData += uart.read(1) # 读取一个字节数据
rxData = uart.readline() # 读取一行数据
uart.readinto(rxbuf, 10) # 读取10个字节数据到rxbuf
uart.sendbreak() # 中断串口数据发送

txData = b’hello world!’
uart.write(txData) # 串口写入数据
uart.txdone() # 串口数据 True:发送完成, False:还有数据未发送成功
uart.flush() # 串口刷新数据
uart.deinit() # 串口反初始化

uart.CTS
uart.RTS
uart.INV_RX
uart.INV_TX

from machine import Pin,UART
import utime

uart0 = UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=Pin(16), rx=Pin(17), )
#uart0.deinit()

txData = b'hello world\n\r'
rxData = bytes()
uart0.write(txData)
#buf=bytes()
#uart0.readinto(buf, 10)
#uart0.sendbreak()

print(type(txData))
print(type(rxData))
while True:
    while uart0.any() > 0:
        #rxData += uart0.read(1)
        rxData = uart0.readline()
        print(rxData.decode('utf-8'))
        if not uart0.txdone():
            uart0.flush()
        uart0.write(rxData)
    utime.sleep(0.2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

Timer定时器

初始化并启动定时器：

• Timer.init(*, mode=Timer.PERIODIC, period=- 1, callback=None)
  mode: 定时器模式(Timer.ONE_SHOT , Timer.PERIODIC)
  period:定时器周期
  callback:定时器回调函数
• Timer.deinit()

启动一个定时器：

from machine import Pin, Timer

led = Pin(25, Pin.OUT)
tim = Timer()

def tick(timer):
   global led
   led.toggle()

tim.init(freq=2.5, mode=Timer.PERIODIC, callback=tick)
1
2
3
4
5
6
7
8
9
10

启动多个定时器：

import utime
from machine import Timer

def tim_callback1(timer):
    """
    Timer Callback Function.
    """
    print("Timer 1...")
    pass

def tim_callback2(timer):
    print("Timer 2...")
    pass

tim1 = Timer()
tim1.init(mode=Timer.PERIODIC, period=1000, callback=tim_callback1)
#tim.init(mode=Timer.ONE_SHOT, period=100, callback=tim_callback)

tim2 = Timer()
tim2.init(mode=Timer.PERIODIC, period=2000, callback=tim_callback2)

print("timer demo ...")

#tim1.deinit()
#tim2.deinit()
while True:
    utime.sleep(1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

双核多线程

RP2040 CPU是双核默认代码运行在core0通过
_thread.start_new_thread启动的线程代码运行在core1上，注：start_new_thread只能调用一次，多次调用会提示core 1已使用；

import time, _thread

def th1_task(n, delay):
    while True:
        print("thread 1 running...",n)
        time.sleep(delay)
    pass

_thread.start_new_thread(th1_task,(10, 1))

print("thread demo...")
while True:
    print("main loop...")
    time.sleep(10)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

使用线程互斥锁：

import utime
import _thread

# 线程互斥锁
lock = _thread.allocate_lock()

# 内核1 运行任务
def core1_task():
    """
    Core1 task
    """
    while True:
        #lock.acquire()
        print("core1 task")
        utime.sleep(0.5)
        #lock.release()
    pass

_thread.start_new_thread(core1_task, ())

while True:
    #lock.acquire()
    print("core0 task")
    utime.sleep(0.5)
    #lock.release()
        
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

文件操作

file = open("test.txt", "w")
file.write("Hello, File!")
file.write(str('13245')+"\n")
file.flush()
file.close()

file = open("test.txt")
print(file.read())
file.close()
1
2
3
4
5
6
7
8
9

模块

超声波测距模块

import machine
import utime

trig = machine.Pin(16, machine.Pin.OUT)
echo = machine.Pin(17, machine.Pin.IN)

def get_distance(trig, echo):
    # 10us high
    trig.low()
    utime.sleep_us(2)
    trig.high()
    utime.sleep_us(10)
    trig.low()
    
    while (echo.value() == 0):
        start = utime.ticks_us()
    while echo.value() == 1:
        end = utime.ticks_us()
        d = (end - start) * 0.034 / 2
    return d

trig.low()

while True:
    distance = get_distance(trig, echo)
    print("距离:{:.2f} cm".format(distance))
    utime.sleep(0.1)
    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

舵机控制

棕色：GND（电源负极）
红色：VCC（电源正极）
橙色：PWM（控制信号）

控制时序0~180角度

控制时序-90 ~ 0 ~ 90角度

from machine import Pin, PWM
import utime

pwm0 = PWM(Pin(0))
pwm0.freq(50)
updown = True

# 舵机控制信号周期T=20ms, f=1/T = 1/20ms = 50Hz
# 脉冲宽度：0.5ms ---> 0度    PWM_DUTY=65535*(0.5ms/20ms)=1638
# 脉冲宽度：1.5ms ---> 90度   PWM_DUTY=65535*(1.5ms/20ms)=4915
# 脉冲宽度：2.5ms ---> 180度  PWM_DUTY=65535*(2.5ms/20ms)=8192

vmin, vmax = (65535*(0.5/20)), (65535*(2.5/20))
def get_angle(angle):   
    duty = int(vmin + (((vmax-vmin)/180) *angle) )
    return duty

angle = 0
pwm0.duty_u16(get_angle(angle))
utime.sleep(2)

while True:
    angle = 0
    pwm0.duty_u16(get_angle(angle))
    utime.sleep(1)
    angle = 45
    pwm0.duty_u16(get_angle(angle))
    utime.sleep(1)
    angle = 90
    pwm0.duty_u16(get_angle(angle))
    utime.sleep(1)
    angle = 135
    pwm0.duty_u16(get_angle(angle))
    utime.sleep(1)
    angle = 180
    pwm0.duty_u16(get_angle(angle))
    utime.sleep(1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

WS2812 RGB灯

将该文件通过Thonny保存到Raspberry PI Pico 的：“/lib/ws2812b.py”目录下


文件已保存到Raspberry Pi Pico内，然后就可以在代码使用import ws2812b

• ws2812b(num, sm, pin)
  num代表ws2812的数量
  sm是内核，目前需要设置为0
  pin是使用的引脚
• set_pixel(n, r, g, b)
  n是第几个ws2812
  r, b, b是红绿蓝颜色
• show()，刷新显示
• fill((r, g, b))，填充所有ws2812
• set_pixel_line(n1,n2,r,g,b)，设置从n1到n2颜色
• set_pixel_line_gradient(n1,n2,r1,g1,b1,r2,g2,b2)，设置从n1到n2渐变色
  ws2812b.py

import array, time
from machine import Pin
import rp2

@rp2.asm_pio(sideset_init=rp2.PIO.OUT_LOW, out_shiftdir=rp2.PIO.SHIFT_LEFT, autopull=True, pull_thresh=24)
def ws2812():
    T1 = 2
    T2 = 5
    T3 = 3
    wrap_target()
    label("bitloop")
    out(x, 1)               .side(0)    [T3 - 1]
    jmp(not_x, "do_zero")   .side(1)    [T1 - 1]
    jmp("bitloop")          .side(1)    [T2 - 1]
    label("do_zero")
    nop()                   .side(0)    [T2 - 1]
    wrap()
    
#delay here is the reset time. You need a pause to reset the LED strip back to the initial LED
#however, if you have quite a bit of processing to do before the next time you update the strip
#you could put in delay=0 (or a lower delay)
class ws2812b:
    def __init__(self, num_leds, state_machine, pin, delay=0.001):
        self.pixels = array.array("I", [0 for _ in range(num_leds)])
        self.sm = rp2.StateMachine(state_machine, ws2812, freq=8000000, sideset_base=Pin(pin))
        self.sm.active(1)
        self.num_leds = num_leds
        self.delay = delay
        self.brightnessvalue = 255

    # Set the overal value to adjust brightness when updating leds
    def brightness(self, brightness = None):
        if brightness == None:
            return self.brightnessvalue
        else:
            if (brightness < 1):
                brightness = 1
        if (brightness > 255):
            brightness = 255
        self.brightnessvalue = brightness

      # Create a gradient with two RGB colors between "pixel1" and "pixel2" (inclusive)
    def set_pixel_line_gradient(self, pixel1, pixel2, left_red, left_green, left_blue, right_red, right_green, right_blue):
        if pixel2 - pixel1 == 0: return
    
        right_pixel = max(pixel1, pixel2)
        left_pixel = min(pixel1, pixel2)
        
        for i in range(right_pixel - left_pixel + 1):
            fraction = i / (right_pixel - left_pixel)
            red = round((right_red - left_red) * fraction + left_red)
            green = round((right_green - left_green) * fraction + left_green)
            blue = round((right_blue - left_blue) * fraction + left_blue)
            
            self.set_pixel(left_pixel + i, red, green, blue)
    
      # Set an array of pixels starting from "pixel1" to "pixel2" to the desired color.
    def set_pixel_line(self, pixel1, pixel2, red, green, blue):
        for i in range(pixel1, pixel2+1):
            self.set_pixel(i, red, green, blue)

    def set_pixel(self, pixel_num, red, green, blue):
        # Adjust color values with brightnesslevel
        blue = round(blue * (self.brightness() / 255))
        red = round(red * (self.brightness() / 255))
        green = round(green * (self.brightness() / 255))

        self.pixels[pixel_num] = blue | red << 8 | green << 16
    
    # rotate x pixels to the left
    def rotate_left(self, num_of_pixels):
        if num_of_pixels == None:
            num_of_pixels = 1
        self.pixels = self.pixels[num_of_pixels:] + self.pixels[:num_of_pixels]

    # rotate x pixels to the right
    def rotate_right(self, num_of_pixels):
        if num_of_pixels == None:
            num_of_pixels = 1
        num_of_pixels = -1 * num_of_pixels
        self.pixels = self.pixels[num_of_pixels:] + self.pixels[:num_of_pixels]

    def show(self):
        for i in range(self.num_leds):
            self.sm.put(self.pixels[i],8)
        time.sleep(self.delay)
            
    def fill(self, red, green, blue):
        for i in range(self.num_leds):
            self.set_pixel(i, red, green, blue)
        time.sleep(self.delay)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

Demo演示代码

from machine  import Pin
import utime
import ws2812b

print(str(ws2812b))

led = ws2812b.ws2812b(9, 0, 6)	# 参数1：LED灯个数， 参数2:固定填0，参数3：Pin端口号

led.fill(255, 0, 0)
led.set_pixel(0, 255, 0, 0) # 设置第一个像素，为红色
#led.set_pixel_line(1, 3, 255, 255, 0)
led.set_pixel_line_gradient(0, 8, 255, 0, 0, 0, 255, 0)# 设置渐变
led.show()

while True:
    #led.rotate_left(1)
    led.rotate_right(1)
    led.show()
    utime.sleep(1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

SSD1306 OLED屏

ssd1306.py保存到pico的/lib目录

# MicroPython SSD1306 OLED driver, I2C and SPI interfaces

from micropython import const
import framebuf


# register definitions
SET_CONTRAST = const(0x81)
SET_ENTIRE_ON = const(0xA4)
SET_NORM_INV = const(0xA6)
SET_DISP = const(0xAE)
SET_MEM_ADDR = const(0x20)
SET_COL_ADDR = const(0x21)
SET_PAGE_ADDR = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP = const(0xA0)
SET_MUX_RATIO = const(0xA8)
SET_COM_OUT_DIR = const(0xC0)
SET_DISP_OFFSET = const(0xD3)
SET_COM_PIN_CFG = const(0xDA)
SET_DISP_CLK_DIV = const(0xD5)
SET_PRECHARGE = const(0xD9)
SET_VCOM_DESEL = const(0xDB)
SET_CHARGE_PUMP = const(0x8D)

# Subclassing FrameBuffer provides support for graphics primitives
# http://docs.micropython.org/en/latest/pyboard/library/framebuf.html
class SSD1306(framebuf.FrameBuffer):
    def __init__(self, width, height, external_vcc):
        self.width = width
        self.height = height
        self.external_vcc = external_vcc
        self.pages = self.height // 8
        self.buffer = bytearray(self.pages * self.width)
        super().__init__(self.buffer, self.width, self.height, framebuf.MONO_VLSB)
        self.init_display()

    def init_display(self):
        for cmd in (
            SET_DISP | 0x00,  # off
            # address setting
            SET_MEM_ADDR,
            0x00,  # horizontal
            # resolution and layout
            SET_DISP_START_LINE | 0x00,
            SET_SEG_REMAP | 0x01,  # column addr 127 mapped to SEG0
            SET_MUX_RATIO,
            self.height - 1,
            SET_COM_OUT_DIR | 0x08,  # scan from COM[N] to COM0
            SET_DISP_OFFSET,
            0x00,
            SET_COM_PIN_CFG,
            0x02 if self.width > 2 * self.height else 0x12,
            # timing and driving scheme
            SET_DISP_CLK_DIV,
            0x80,
            SET_PRECHARGE,
            0x22 if self.external_vcc else 0xF1,
            SET_VCOM_DESEL,
            0x30,  # 0.83*Vcc
            # display
            SET_CONTRAST,
            0xFF,  # maximum
            SET_ENTIRE_ON,  # output follows RAM contents
            SET_NORM_INV,  # not inverted
            # charge pump
            SET_CHARGE_PUMP,
            0x10 if self.external_vcc else 0x14,
            SET_DISP | 0x01,
        ):  # on
            self.write_cmd(cmd)
        self.fill(0)
        self.show()

    def poweroff(self):
        self.write_cmd(SET_DISP | 0x00)

    def poweron(self):
        self.write_cmd(SET_DISP | 0x01)

    def contrast(self, contrast):
        self.write_cmd(SET_CONTRAST)
        self.write_cmd(contrast)

    def invert(self, invert):
        self.write_cmd(SET_NORM_INV | (invert & 1))

    def show(self):
        x0 = 0
        x1 = self.width - 1
        if self.width == 64:
            # displays with width of 64 pixels are shifted by 32
            x0 += 32
            x1 += 32
        self.write_cmd(SET_COL_ADDR)
        self.write_cmd(x0)
        self.write_cmd(x1)
        self.write_cmd(SET_PAGE_ADDR)
        self.write_cmd(0)
        self.write_cmd(self.pages - 1)
        self.write_data(self.buffer)


class SSD1306_I2C(SSD1306):
    def __init__(self, width, height, i2c, addr=0x3C, external_vcc=False):
        self.i2c = i2c
        self.addr = addr
        self.temp = bytearray(2)
        self.write_list = [b"\x40", None]  # Co=0, D/C#=1
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.temp[0] = 0x80  # Co=1, D/C#=0
        self.temp[1] = cmd
        self.i2c.writeto(self.addr, self.temp)

    def write_data(self, buf):
        self.write_list[1] = buf
        self.i2c.writevto(self.addr, self.write_list)


class SSD1306_SPI(SSD1306):
    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
        self.rate = 10 * 1024 * 1024
        dc.init(dc.OUT, value=0)
        res.init(res.OUT, value=0)
        cs.init(cs.OUT, value=1)
        self.spi = spi
        self.dc = dc
        self.res = res
        self.cs = cs
        import time

        self.res(1)
        time.sleep_ms(1)
        self.res(0)
        time.sleep_ms(10)
        self.res(1)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs(1)
        self.dc(0)
        self.cs(0)
        self.spi.write(bytearray([cmd]))
        self.cs(1)

    def write_data(self, buf):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs(1)
        self.dc(1)
        self.cs(0)
        self.spi.write(buf)
        self.cs(1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

from machine import Pin,I2C
from ssd1306 import SSD1306_I2C
import framebuf
import utime
import array

i2c = I2C(0, scl=Pin(9), sda=Pin(8))
print("I2C Address:", hex(i2c.scan()[0]).upper())

WIDTH  = 128
HEIGHT = 64
oled = SSD1306_I2C(WIDTH, HEIGHT, i2c)

# Raspberry Pi logo as 32x32 bytearray
buffer = bytearray(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00|?\x00\x01\x86@\x80\x01\x01\x80\x80\x01\x11\x88\x80\x01\x05\xa0\x80\x00\x83\xc1\x00\x00C\xe3\x00\x00~\xfc\x00\x00L'\x00\x00\x9c\x11\x00\x00\xbf\xfd\x00\x00\xe1\x87\x00\x01\xc1\x83\x80\x02A\x82@\x02A\x82@\x02\xc1\xc2@\x02\xf6>\xc0\x01\xfc=\x80\x01\x18\x18\x80\x01\x88\x10\x80\x00\x8c!\x00\x00\x87\xf1\x00\x00\x7f\xf6\x00\x008\x1c\x00\x00\x0c \x00\x00\x03\xc0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")

# Load the raspberry pi logo into the framebuffer (the image is 32x32)
fb = framebuf.FrameBuffer(buffer, 32, 32, framebuf.MONO_HLSB)


oled.fill(0)
oled.blit(fb, (128-32)//2, (64-32)//2)
oled.text('RaspberryPi Pico', 0, 48)
oled.show()
utime.sleep(1)

#oled.invert(True)	# 反色
#oled.invert(False) # 正常
#oled.poweroff()
#oled.poweron()
oled.contrast(250) # 0x0~0xFF

oled.fill(0)
oled.text("Hello", 15, 30)
oled.text("RP2040", 15, 40)
oled.hline(0,0,60, 1)
oled.hline(0,20,60, 1)  # 水平线， X,Y, W, 颜色
oled.vline(0,0,20, 1)
oled.vline(60,0,20, 1)  # 垂直线， X,Y, W, 颜色
oled.rect(80, 0, 40, 10, 1, True) # X,Y, 宽度，高度，颜色，True:实心矩形，False:空心矩形
oled.ellipse(40, 20, 30, 10, 1, True, 0b0010) # 中心X，中心Y，宽度，高度，颜色，实心/空心  象限Q4|Q3|Q2|Q1
oled.pixel(64, 32, 1)   # 画一个像素，X,Y, 颜色

coords = array.array('h', [0,0,  15,12, 2,8,  16,26]) # 路径坐标， [x0,y0, x1,y1,...xn,yn]
oled.poly(60, 20, coords, 1, False) #多边形  x, y, 路径，颜色，实心/空心
oled.show()


while True:
#     oled.scroll(0,1)
    oled.show()
    utime.sleep_ms(200)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

AHT10温湿度传感器

aht10.py另存到pico的/lib/aht10.py

import time
from math import log

# AHT10 Library for MicroPython on ESP32
# Author: Sean Yong
# Date: 23rd December, 2019
# Version 1.0

#CONSTANTS
AHT10_ADDRESS = 0x38 # 0111000 (7bit address)
AHT10_READ_DELAY_MS = 75 # Time it takes for AHT to collect data
AHT_TEMPERATURE_CONST = 200
AHT_TEMPERATURE_OFFSET = 50
KILOBYTE_CONST = 1048576
CMD_INITIALIZE = bytearray([0xE1, 0x08, 0x00])
CMD_MEASURE = bytearray([0xAC, 0x33, 0x00])
FARENHEIT_MULTIPLIER = 9/5
FARENHEIT_OFFSET = 32

class AHT10:
    def __init__(self, i2c, mode=0, address=AHT10_ADDRESS):
        if i2c is None:
            raise ValueError('I2C object required.')
        if mode is not (0 and 1):
            raise ValueError('Mode must be either 0 for Celsius or 1 Farenheit')
        self.i2c = i2c
        self.address = address
        self.i2c.writeto(address, CMD_INITIALIZE)
        self.readings_raw = bytearray(8)
        self.results_parsed = [0, 0]
        self.mode = mode # 0 for Celsius, 1 for Farenheit

    def read_raw(self):
        self.i2c.writeto(self.address, CMD_MEASURE)
        time.sleep_ms(AHT10_READ_DELAY_MS)
        self.readings_raw = self.i2c.readfrom(AHT10_ADDRESS, 6)
        self.results_parsed[0] = self.readings_raw[1] << 12 | self.readings_raw[2] << 4 | self.readings_raw[3] >> 4
        self.results_parsed[1] = (self.readings_raw[3] & 0x0F) << 16 | self.readings_raw[4] << 8 | self.readings_raw[5]

    def humidity(self):
        self.read_raw()
        return (self.results_parsed[0] / KILOBYTE_CONST) * 100 

    def temperature(self):
        self.read_raw()
        if self.mode is 0:
            return (self.results_parsed[1] / KILOBYTE_CONST) * AHT_TEMPERATURE_CONST - AHT_TEMPERATURE_OFFSET
        else:
            return ((self.results_parsed[1] / KILOBYTE_CONST) * AHT_TEMPERATURE_CONST - AHT_TEMPERATURE_OFFSET) * FARENHEIT_MULTIPLIER + FARENHEIT_OFFSET

    def set_mode(self, mode):
        if mode is not (0 or 1):
            raise ValueError('Mode must be either 0 for Celsius or 1 Farenheit')
        self.mode = mode

    def print(self):
        print("Temperature: " + str(self.temperature()) + ("C","F")[self.mode] + ", Humidity: " + str(self.humidity()))

    def dew_point(self):
        h = self.humidity()
        t = self.temperature()
        prev_mode = self.mode
        self.mode = 0
        h = (log(h, 10) - 2) / 0.4343 + (17.62 * t) / (243.12 + t)
        return 243.12 * h / (17.62 - h)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

from machine import Pin,I2C
import utime
import aht10

i2c = I2C(0, sda=Pin(8), scl=Pin(9),freq=400000)
devlist = i2c.scan()
print("Address List:",devlist)
for dev in devlist:
    print(hex(dev))

aht10 = aht10.AHT10(i2c, mode=0, address=0x38)

while True:
    humi = aht10.humidity()     # 湿度
    temp = aht10.temperature()  # 温度
    dewp = aht10.dew_point()    # 露点
    aht10.print()
    utime.sleep(1)    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

OLED显示AHT10读取值

from machine import Pin,I2C
from ssd1306 import SSD1306_I2C,SSD1306_SPI
import framebuf
import array
import utime
import aht10


i2c = I2C(0, sda=Pin(8), scl=Pin(9),freq=400000)
devlist = i2c.scan()
print("Address List:",devlist)
for dev in devlist:
    print(hex(dev))



# Raspberry Pi logo as 32x32 bytearray
buffer = bytearray(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00|?\x00\x01\x86@\x80\x01\x01\x80\x80\x01\x11\x88\x80\x01\x05\xa0\x80\x00\x83\xc1\x00\x00C\xe3\x00\x00~\xfc\x00\x00L'\x00\x00\x9c\x11\x00\x00\xbf\xfd\x00\x00\xe1\x87\x00\x01\xc1\x83\x80\x02A\x82@\x02A\x82@\x02\xc1\xc2@\x02\xf6>\xc0\x01\xfc=\x80\x01\x18\x18\x80\x01\x88\x10\x80\x00\x8c!\x00\x00\x87\xf1\x00\x00\x7f\xf6\x00\x008\x1c\x00\x00\x0c \x00\x00\x03\xc0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")

# Load the raspberry pi logo into the framebuffer (the image is 32x32)
fb = framebuf.FrameBuffer(buffer, 32, 32, framebuf.MONO_HLSB)

WIDTH = 128
HEIGHT = 64
oled = SSD1306_I2C(WIDTH,HEIGHT,i2c)
oled.fill(0)
# oled.blit(fb, (128-32)//2, (64-32)//2)
oled.text('RaspberryPi Pico', 0, 0)
oled.hline(0,12, 128, 1)
oled.blit(fb, (128-32), 15)
oled.show()

aht10 = aht10.AHT10(i2c, mode=0, address=0x38)
utime.sleep(1)

while True:
    humi = aht10.humidity()
    temp = aht10.temperature()
    dewp = aht10.dew_point()
    #aht10.print()
    utime.sleep(0.2)
    #oled.fill(0)
    oled.rect(0,15,128-32,45, 0, True)
    oled.text("Temp:{:.2f}C".format(temp), 0, 15)
    oled.text("Humi:{:.2f}%".format(humi), 0, 25)
    oled.text("Dewp:{:.2f}C".format(dewp), 0, 35)
    oled.show()
    
    utime.sleep(1)    
  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

BH1750光照度传感器

bh1750.py

import utime

# BH1750 Library for MicroPython on RP2040
# Author: Song YuLong
# Date: 2022-10-25
# Version 1.0

#CONSTANTS
BH1750_ADDRESS = const (0x23) # 00100011 I2C Slave Address
BH1750_READ_DELAY_MS = 180    # measurement time ms.

BH1750_POWERDOWN =  bytearray([0x00])		# Power Down
BH1750_POWERON   =  bytearray([0x01])		# Power On
BH1750_RESET     =  bytearray([0x07])		# Reset Data Register Value.

# Countinue :
# One Time  : it is automatically set to power down mode after measurement.
BH1750_CONTINUE_H_MODE  = bytearray([0x10]) # Countinue H-Resolution Mode 1lx 120ms
BH1750_CONTINUE_H_MODE2 = bytearray([0x11]) # Countinue H-Resolution Mode 0.5lx 120ms
BH1750_CONTINUE_L_MODE  = bytearray([0x13])	# Countinue L-Resolution Mode 4lx 16ms
BH1750_ONE_TIME_H_MODE  = bytearray([0x20]) # One Time H-Resolution Mode 1lx 120ms
BH1750_ONE_TIME_H_MODE2 = bytearray([0x21]) # One Time H-Resolution Mode 0.5lx 120ms
BH1750_ONE_TIME_L_MODE  = bytearray([0x23]) # One Time L-Resolution Mode 4lx 16ms

class BH1750:
    def __init__(self, i2c, mode=0, address=BH1750_ADDRESS):
        if i2c is None:
            raise ValueError('I2C object required.')
        
        if mode == 0:
            self.mode = BH1750_CONTINUE_H_MODE
        elif mode == 1:
            self.mode = BH1750_CONTINUE_H_MODE2
        elif mode == 2:
            self.mode = BH1750_CONTINUE_L_MODE
        elif mode == 3:
            self.mode = BH1750_ONE_TIME_H_MODE
        elif mode == 4:
            self.mode = BH1750_ONE_TIME_H_MODE2
        elif mode == 5:
            self.mode = BH1750_ONE_TIME_L_MODE
        
        if self.mode == BH1750_CONTINUE_H_MODE2 or self.mode == BH1750_ONE_TIME_H_MODE2:
            self.resolurtion = 0.5
        elif self.mode == BH1750_CONTINUE_H_MODE or self.mode == BH1750_ONE_TIME_H_MODE:
            self.resolurtion = 1
        elif self.mode == BH1750_CONTINUE_L_MODE or self.mode == BH1750_ONE_TIME_L_MODE:
            self.resolurtion = 4
        else:
            raise ValueError('Mode must be 0 ~ 5')
        
        self.i2c = i2c
        self.address = address
        self.i2c.writeto(address, BH1750_POWERDOWN)
        self.i2c.writeto(address, BH1750_POWERON)
        self.i2c.writeto(address, self.mode)
        self.rawdata = bytearray(8)
        self.value = 0.0
        
    def read_raw(self):
        self.i2c.writeto(self.address, self.mode)
        utime.sleep_ms(BH1750_READ_DELAY_MS) # max 160ms
        self.rawdata = self.i2c.readfrom(self.address, 2) 
    
    def illuminance(self):
        # illumination =(reg[15:0] * resolurtion) / 1.2 （unit:lx）
        self.read_raw()
        self.value = ((self.rawdata[0]<<8 | self.rawdata[1]) * self.resolurtion)/1.2
        return self.value
    
    def output(self):
        print("illuminance -->{:.2f}".format(self.value), "LX")
    
    def set_mode(self, mode):
        if mode == 0:
            self.mode = BH1750_CONTINUE_H_MODE
        elif mode == 1:
            self.mode = BH1750_CONTINUE_H_MODE2
        elif mode == 2:
            self.mode = BH1750_CONTINUE_L_MODE
        elif mode == 3:
            self.mode = BH1750_ONE_TIME_H_MODE
        elif mode == 4:
            self.mode = BH1750_ONE_TIME_H_MODE2
        elif mode == 5:
            self.mode = BH1750_ONE_TIME_L_MODE
            
        if self.mode == BH1750_CONTINUE_H_MODE2 or self.mode == BH1750_ONE_TIME_H_MODE2:
            self.resolurtion = 0.5
        elif self.mode == BH1750_CONTINUE_H_MODE or self.mode == BH1750_ONE_TIME_H_MODE:
            self.resolurtion = 1
        elif self.mode == BH1750_CONTINUE_L_MODE or self.mode == BH1750_ONE_TIME_L_MODE:
            self.resolurtion = 4
        else:
            raise ValueError('Mode must be 0 ~ 5')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

bh1750_demo.py

from machine import Pin,I2C
import utime
import bh1750

i2c = I2C(0, sda=Pin(8), scl=Pin(9), freq=200000)
devlist = i2c.scan()
print(devlist)
for dev in devlist:
    print(hex(dev))

bh1750 = bh1750.BH1750(i2c, mode=1, address=0x23)

print("BH1750 Demo...")
value = 0.0

while True:
    value = bh1750.illuminance()
    print("illuminance:{:.2f} lx".format(value))
    #bh1750.output()
    utime.sleep(0.1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

BMP280 气压强传感器

bmp280.py

from micropython import const
from ustruct import unpack as unp

# Author David Stenwall Wahlund (david at dafnet.se)

# Power Modes
BMP280_POWER_SLEEP = const(0)
BMP280_POWER_FORCED = const(1)
BMP280_POWER_NORMAL = const(3)

BMP280_SPI3W_ON = const(1)
BMP280_SPI3W_OFF = const(0)

BMP280_TEMP_OS_SKIP = const(0)
BMP280_TEMP_OS_1 = const(1)
BMP280_TEMP_OS_2 = const(2)
BMP280_TEMP_OS_4 = const(3)
BMP280_TEMP_OS_8 = const(4)
BMP280_TEMP_OS_16 = const(5)

BMP280_PRES_OS_SKIP = const(0)
BMP280_PRES_OS_1 = const(1)
BMP280_PRES_OS_2 = const(2)
BMP280_PRES_OS_4 = const(3)
BMP280_PRES_OS_8 = const(4)
BMP280_PRES_OS_16 = const(5)

# Standby settings in ms
BMP280_STANDBY_0_5 = const(0)
BMP280_STANDBY_62_5 = const(1)
BMP280_STANDBY_125 = const(2)
BMP280_STANDBY_250 = const(3)
BMP280_STANDBY_500 = const(4)
BMP280_STANDBY_1000 = const(5)
BMP280_STANDBY_2000 = const(6)
BMP280_STANDBY_4000 = const(7)

# IIR Filter setting
BMP280_IIR_FILTER_OFF = const(0)
BMP280_IIR_FILTER_2 = const(1)
BMP280_IIR_FILTER_4 = const(2)
BMP280_IIR_FILTER_8 = const(3)
BMP280_IIR_FILTER_16 = const(4)

# Oversampling setting
BMP280_OS_ULTRALOW = const(0)
BMP280_OS_LOW = const(1)
BMP280_OS_STANDARD = const(2)
BMP280_OS_HIGH = const(3)
BMP280_OS_ULTRAHIGH = const(4)

# Oversampling matrix
# (PRESS_OS, TEMP_OS, sample time in ms)
_BMP280_OS_MATRIX = [
    [BMP280_PRES_OS_1, BMP280_TEMP_OS_1, 7],
    [BMP280_PRES_OS_2, BMP280_TEMP_OS_1, 9],
    [BMP280_PRES_OS_4, BMP280_TEMP_OS_1, 14],
    [BMP280_PRES_OS_8, BMP280_TEMP_OS_1, 23],
    [BMP280_PRES_OS_16, BMP280_TEMP_OS_2, 44]
]

# Use cases
BMP280_CASE_HANDHELD_LOW = const(0)
BMP280_CASE_HANDHELD_DYN = const(1)
BMP280_CASE_WEATHER = const(2)
BMP280_CASE_FLOOR = const(3)
BMP280_CASE_DROP = const(4)
BMP280_CASE_INDOOR = const(5)

_BMP280_CASE_MATRIX = [
    [BMP280_POWER_NORMAL, BMP280_OS_ULTRAHIGH, BMP280_IIR_FILTER_4, BMP280_STANDBY_62_5],
    [BMP280_POWER_NORMAL, BMP280_OS_STANDARD, BMP280_IIR_FILTER_16, BMP280_STANDBY_0_5],
    [BMP280_POWER_FORCED, BMP280_OS_ULTRALOW, BMP280_IIR_FILTER_OFF, BMP280_STANDBY_0_5],
    [BMP280_POWER_NORMAL, BMP280_OS_STANDARD, BMP280_IIR_FILTER_4, BMP280_STANDBY_125],
    [BMP280_POWER_NORMAL, BMP280_OS_LOW, BMP280_IIR_FILTER_OFF, BMP280_STANDBY_0_5],
    [BMP280_POWER_NORMAL, BMP280_OS_ULTRAHIGH, BMP280_IIR_FILTER_16, BMP280_STANDBY_0_5]
]

_BMP280_REGISTER_ID = const(0xD0)
_BMP280_REGISTER_RESET = const(0xE0)
_BMP280_REGISTER_STATUS = const(0xF3)
_BMP280_REGISTER_CONTROL = const(0xF4)
_BMP280_REGISTER_CONFIG = const(0xF5)  # IIR filter config

_BMP280_REGISTER_DATA = const(0xF7)


class BMP280:
    def __init__(self, i2c_bus, addr=0x76, use_case=BMP280_CASE_HANDHELD_DYN):
        self._bmp_i2c = i2c_bus
        self._i2c_addr = addr

        # read calibration data
        # < little-endian
        # H unsigned short
        # h signed short
        self._T1 = unp('<H', self._read(0x88, 2))[0]
        self._T2 = unp('<h', self._read(0x8A, 2))[0]
        self._T3 = unp('<h', self._read(0x8C, 2))[0]
        self._P1 = unp('<H', self._read(0x8E, 2))[0]
        self._P2 = unp('<h', self._read(0x90, 2))[0]
        self._P3 = unp('<h', self._read(0x92, 2))[0]
        self._P4 = unp('<h', self._read(0x94, 2))[0]
        self._P5 = unp('<h', self._read(0x96, 2))[0]
        self._P6 = unp('<h', self._read(0x98, 2))[0]
        self._P7 = unp('<h', self._read(0x9A, 2))[0]
        self._P8 = unp('<h', self._read(0x9C, 2))[0]
        self._P9 = unp('<h', self._read(0x9E, 2))[0]

        # output raw
        self._t_raw = 0
        self._t_fine = 0
        self._t = 0

        self._p_raw = 0
        self._p = 0
        
        self._h = 0

        self.read_wait_ms = 0  # interval between forced measure and readout
        self._new_read_ms = 200  # interval between
        self._last_read_ts = 0

        if use_case is not None:
            self.use_case(use_case)

    def _read(self, addr, size=1):
        return self._bmp_i2c.readfrom_mem(self._i2c_addr, addr, size)

    def _write(self, addr, b_arr):
        if not type(b_arr) is bytearray:
            b_arr = bytearray([b_arr])
        return self._bmp_i2c.writeto_mem(self._i2c_addr, addr, b_arr)

    def _gauge(self):
        # TODO limit new reads
        # read all data at once (as by spec)
        d = self._read(_BMP280_REGISTER_DATA, 6)

        self._p_raw = (d[0] << 12) + (d[1] << 4) + (d[2] >> 4)
        self._t_raw = (d[3] << 12) + (d[4] << 4) + (d[5] >> 4)

        self._t_fine = 0
        self._t = 0
        self._p = 0
        self._h = 0

    def reset(self):
        self._write(_BMP280_REGISTER_RESET, 0xB6)

    def load_test_calibration(self):
        self._T1 = 27504
        self._T2 = 26435
        self._T3 = -1000
        self._P1 = 36477
        self._P2 = -10685
        self._P3 = 3024
        self._P4 = 2855
        self._P5 = 140
        self._P6 = -7
        self._P7 = 15500
        self._P8 = -14600
        self._P9 = 6000

    def load_test_data(self):
        self._t_raw = 519888
        self._p_raw = 415148

    def print_calibration(self):
        print("T1: {} {}".format(self._T1, type(self._T1)))
        print("T2: {} {}".format(self._T2, type(self._T2)))
        print("T3: {} {}".format(self._T3, type(self._T3)))
        print("P1: {} {}".format(self._P1, type(self._P1)))
        print("P2: {} {}".format(self._P2, type(self._P2)))
        print("P3: {} {}".format(self._P3, type(self._P3)))
        print("P4: {} {}".format(self._P4, type(self._P4)))
        print("P5: {} {}".format(self._P5, type(self._P5)))
        print("P6: {} {}".format(self._P6, type(self._P6)))
        print("P7: {} {}".format(self._P7, type(self._P7)))
        print("P8: {} {}".format(self._P8, type(self._P8)))
        print("P9: {} {}".format(self._P9, type(self._P9)))

    def _calc_t_fine(self):
        # From datasheet page 22
        self._gauge()
        if self._t_fine == 0:
            var1 = (((self._t_raw >> 3) - (self._T1 << 1)) * self._T2) >> 11
            var2 = (((((self._t_raw >> 4) - self._T1)
                      * ((self._t_raw >> 4)
                         - self._T1)) >> 12)
                    * self._T3) >> 14
            self._t_fine = var1 + var2

    @property
    def temperature(self):
        self._calc_t_fine()
        if self._t == 0:
            self._t = ((self._t_fine * 5 + 128) >> 8) / 100.
        return self._t

    @property
    def pressure(self):
        # From datasheet page 22
        self._calc_t_fine()
        if self._p == 0:
            var1 = self._t_fine - 128000
            var2 = var1 * var1 * self._P6
            var2 = var2 + ((var1 * self._P5) << 17)
            var2 = var2 + (self._P4 << 35)
            var1 = ((var1 * var1 * self._P3) >> 8) + ((var1 * self._P2) << 12)
            var1 = (((1 << 47) + var1) * self._P1) >> 33

            if var1 == 0:
                return 0

            p = 1048576 - self._p_raw
            p = int((((p << 31) - var2) * 3125) / var1)
            var1 = (self._P9 * (p >> 13) * (p >> 13)) >> 25
            var2 = (self._P8 * p) >> 19

            p = ((p + var1 + var2) >> 8) + (self._P7 << 4)
            self._p = p / 256.0
        return self._p
    
    @property
    def altitude_p_t(self):
        '''
        Calculate altitude from current temperature and pressure.
        h = ( (((P0/P)^(1/5.257))-1) * (T + 237.15) ) / 0.0065
        h:altitude (m)
        P0:101.325(kPa)
        P:current pressure in kPa
        T:current temperature in Celsius
        '''
        self._h = ( (((101.325/(self._p/1000))**(1/5.257))-1) * (self._t + 273.15) ) / 0.0065
        return self._h
    
    @property
    def altitude_p(self):
        self._h = 44330 * (1 - (((self._p/1000)/101.325)**(1/5.255)))
        return self._h
    
    def _write_bits(self, address, value, length, shift=0):
        d = self._read(address)[0]
        m = int('1' * length, 2) << shift
        d &= ~m
        d |= m & value << shift
        self._write(address, d)

    def _read_bits(self, address, length, shift=0):
        d = self._read(address)[0]
        return d >> shift & int('1' * length, 2)

    @property
    def standby(self):
        return self._read_bits(_BMP280_REGISTER_CONFIG, 3, 5)

    @standby.setter
    def standby(self, v):
        assert 0 <= v <= 7
        self._write_bits(_BMP280_REGISTER_CONFIG, v, 3, 5)

    @property
    def iir(self):
        return self._read_bits(_BMP280_REGISTER_CONFIG, 3, 2)

    @iir.setter
    def iir(self, v):
        assert 0 <= v <= 4
        self._write_bits(_BMP280_REGISTER_CONFIG, v, 3, 2)

    @property
    def spi3w(self):
        return self._read_bits(_BMP280_REGISTER_CONFIG, 1)

    @spi3w.setter
    def spi3w(self, v):
        assert v in (0, 1)
        self._write_bits(_BMP280_REGISTER_CONFIG, v, 1)

    @property
    def temp_os(self):
        return self._read_bits(_BMP280_REGISTER_CONTROL, 3, 5)

    @temp_os.setter
    def temp_os(self, v):
        assert 0 <= v <= 5
        self._write_bits(_BMP280_REGISTER_CONTROL, v, 3, 5)

    @property
    def press_os(self):
        return self._read_bits(_BMP280_REGISTER_CONTROL, 3, 2)

    @press_os.setter
    def press_os(self, v):
        assert 0 <= v <= 5
        self._write_bits(_BMP280_REGISTER_CONTROL, v, 3, 2)

    @property
    def power_mode(self):
        return self._read_bits(_BMP280_REGISTER_CONTROL, 2)

    @power_mode.setter
    def power_mode(self, v):
        assert 0 <= v <= 3
        self._write_bits(_BMP280_REGISTER_CONTROL, v, 2)

    @property
    def is_measuring(self):
        return bool(self._read_bits(_BMP280_REGISTER_STATUS, 1, 3))

    @property
    def is_updating(self):
        return bool(self._read_bits(_BMP280_REGISTER_STATUS, 1))

    @property
    def chip_id(self):
        return self._read(_BMP280_REGISTER_ID, 2)

    @property
    def in_normal_mode(self):
        return self.power_mode == BMP280_POWER_NORMAL

    def force_measure(self):
        self.power_mode = BMP280_POWER_FORCED

    def normal_measure(self):
        self.power_mode = BMP280_POWER_NORMAL

    def sleep(self):
        self.power_mode = BMP280_POWER_SLEEP

    def use_case(self, uc):
        assert 0 <= uc <= 5
        pm, oss, iir, sb = _BMP280_CASE_MATRIX[uc]
        p_os, t_os, self.read_wait_ms = _BMP280_OS_MATRIX[oss]
        self._write(_BMP280_REGISTER_CONFIG, (iir << 2) + (sb << 5))
        self._write(_BMP280_REGISTER_CONTROL, pm + (p_os << 2) + (t_os << 5))

    def oversample(self, oss):
        assert 0 <= oss <= 4
        p_os, t_os, self.read_wait_ms = _BMP280_OS_MATRIX[oss]
        self._write_bits(_BMP280_REGISTER_CONTROL, p_os + (t_os << 3), 2)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344

bmp280_demo.py

from machine import Pin, I2C
import utime
import bmp280
import math
from ustruct import unpack as unp

i2c = I2C(0, sda=Pin(8), scl=Pin(9), freq=200000)
devlist = i2c.scan()
print(devlist)
for dev in devlist:
    print(hex(dev))

print("BMP280 Demo...")
bmp280 = bmp280.BMP280(i2c_bus=i2c, addr=0x76)
bmp280.use_case(1)
bmp280.oversample(1)
bmp280.iir = 2
bmp280.temp_os=2
bmp280.press_os=2

bmp280.print_calibration()
print(bmp280.chip_id)

    
while True:
    t = bmp280.temperature
    p = bmp280.pressure
    h = bmp280.altitude_p_t	# calculate altitude from pressure and temperature
    h1 = bmp280.altitude_p  # calculate altitude from pressure
    print("P:{:.2f}(kPa)".format(p/1000),"T:{:.2f}(C)".format(t), "H:{:.2f}(m)".format(h), "H1:{:.2f}(m)".format(h1))
    #print(p, t, h, h1)
    utime.sleep(0.5)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

运行效果：

HMC5883L地磁传感器

IC丝印L883为HMC8553L
IC丝印5883为QMC5883L

库模块/lib/hmc5883l.py

import math
import machine

from ustruct import pack
from array import array


class HMC5883L:
    __gain__ = {
        '0.88': (0 << 5, 0.73),
        '1.3':  (1 << 5, 0.92),
        '1.9':  (2 << 5, 1.22),
        '2.5':  (3 << 5, 1.52),
        '4.0':  (4 << 5, 2.27),
        '4.7':  (5 << 5, 2.56),
        '5.6':  (6 << 5, 3.03),
        '8.1':  (7 << 5, 4.35)
    }

    def __init__(self, i2c, address=30, gauss='1.3', declination=(0, 0)):
        self.i2c = i2c
        self.address = address

        # Initialize sensor.
        #i2c.start()

        # Configuration register A:
        #   0bx11xxxxx  -> 8 samples averaged per measurement
        #   0bxxx100xx  -> 15 Hz, rate at which data is written to output registers
        #   0bxxxxxx00  -> Normal measurement mode
        i2c.writeto_mem(self.address, 0x00, pack('B', 0b111000))

        # Configuration register B:
        reg_value, self.gain = self.__gain__[gauss]
        i2c.writeto_mem(self.address, 0x01, pack('B', reg_value))

        # Set mode register to continuous mode.
        i2c.writeto_mem(self.address, 0x02, pack('B', 0x00))
        #i2c.stop()

        # Convert declination (tuple of degrees and minutes) to radians.
        self.declination = (declination[0] + declination[1] / 60) * math.pi / 180

        # Reserve some memory for the raw xyz measurements.
        self.data = array('B', [0] * 6)

    def read(self):
        data = self.data
        gain = self.gain

        self.i2c.readfrom_mem_into(self.address, 0x03, data)

        x = (data[0] << 8) | data[1]
        z = (data[2] << 8) | data[3]
        y = (data[4] << 8) | data[5]

        x = x - (1 << 16) if x & (1 << 15) else x
        y = y - (1 << 16) if y & (1 << 15) else y
        z = z - (1 << 16) if z & (1 << 15) else z

        x = round(x * gain, 4)
        y = round(y * gain, 4)
        z = round(z * gain, 4)

        return x, y, z

    def heading(self, x, y):
        heading_rad = math.atan2(y, x)
        heading_rad += self.declination

        # Correct reverse heading.
        if heading_rad < 0:
            heading_rad += 2 * math.pi

        # Compensate for wrapping.
        elif heading_rad > 2 * math.pi:
            heading_rad -= 2 * math.pi

        # Convert from radians to degrees.
        heading = heading_rad * 180 / math.pi
        degrees = math.floor(heading)
        minutes = round((heading - degrees) * 60)
        return degrees, minutes

    def format_result(self, x, y, z):
        degrees, minutes = self.heading(x, y)
        return 'X: {:.4f}, Y: {:.4f}, Z: {:.4f}, Heading: {}° {}′ '.format(x, y, z, degrees, minutes)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

hmc5883l_demo.py

from machine import Pin,I2C
import utime
from hmc5883l import HMC5883L
from ssd1306 import SSD1306_I2C,SSD1306_SPI

i2c0 = I2C(0, sda=Pin(8), scl=Pin(9),freq=400000)
devlist = i2c0.scan()
print("I2C0 Address List:",devlist)
for dev in devlist:
    print(hex(dev))
      
mSensor = HMC5883L(i2c0, address=0x1e, gauss='1.3')


while True:
    x, y, z = mSensor.read()
    print(mSensor.format_result(x, y, z))
    utime.sleep(0.1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

MCP4725数模转换

/lib/mcp4725.py

#Library for the MCP4725 I2C bus DAC 
from machine import I2C

#The MCP4725 has support from 2 addresses
BUS_ADDRESS = [0x60,0x62,0x63]

#The device supports a few power down modes on startup and during operation 
POWER_DOWN_MODE = {'Off':0, '1k':1, '100k':2, '500k':3}
        
class MCP4725:
    def __init__(self,i2c, address=BUS_ADDRESS[0]) :
        self.i2c=i2c
        self.address=address
        self._writeBuffer=bytearray(2)
        
    def write(self,value):
        if value < 0:
            value=0
        value=value & 0xFFF
        self._writeBuffer[0]=(value>>8) & 0xFF
        self._writeBuffer[1]=value & 0xFF
        return self.i2c.writeto(self.address,self._writeBuffer)==2

    def read(self):
        buf=bytearray(5)
        if self.i2c.readfrom_into(self.address,buf) ==5:
            eeprom_write_busy=(buf[0] & 0x80)==0
            power_down=self._powerDownKey((buf[0] >> 1) & 0x03)
            value=((buf[1]<<8) | (buf[2])) >> 4
            eeprom_power_down=self._powerDownKey((buf[3]>>5) & 0x03)
            eeprom_value=((buf[3] & 0x0f)<<8) | buf[4] 
            return (eeprom_write_busy,power_down,value,eeprom_power_down,eeprom_value)
        return None

    def config(self,power_down='Off',value=0,eeprom=False):
        buf=bytearray()
        conf=0x40 | (POWER_DOWN_MODE[power_down] << 1)
        if eeprom:
            #store the powerdown and output value in eeprom
            conf=conf | 0x60
        buf.append(conf)
        #check value range
        if value<0:
            value=0
        value=value & 0xFFF
        buf.append(value >> 4)
        buf.append((value & 0x0F)<<4)
        return self.i2c.writeto(self.address,buf)==3

    def _powerDownKey(self,value):
        for key,item in POWER_DOWN_MODE.items():
            if item == value:
                return key

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

mcp4725_demo.py

from machine import Pin,I2C
import mcp4725
import utime
import math
import machine


i2c0 = I2C(0, sda=Pin(8), scl=Pin(9), freq=400000)

devlist = i2c0.scan()
print(devlist)
for dev in devlist:
    print(hex(dev))

#create the MCP4725 driver
dac=mcp4725.MCP4725(i2c0,mcp4725.BUS_ADDRESS[0])

#configure the DAC to output ``Vdd/2`` on power-up or after a reset
#dac.config('Off',2048, eeprom=True)

#configure the DAC to go into power-down mode and set the output value to maximum output.
dac.config('100k',4096)
print(dac.read())

x = 0
value = 0
updown = True

while True:
#     print(value)
    dac.write(value)
    # 正弦波
    value = 2048+int(math.sin(x)*2048)
    x += 0.01
    if x >= 360:
        x = 0.0
    
    
    # 方波
    '''
    if value == 0:
        value = 4095
    else:
        value = 0
    '''
    
    # 锯齿波 正
    '''
    if value < 4096:
        value += 1
    elif value == 4096:
        value=0
    '''
    # 锯齿波 反
    '''
    if value > 0:
        value -= 1
    elif value <= 0:
        value = 4095  
    '''
    # 三角波
    '''
    if updown :
        value+=1
    else:
        value-=1
        
    if value == 4096:
        updown = False
    elif value == 0:
        updown = True
    '''        

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

正弦波

方波

锯齿波


三角波

GNSS导航模块

navigation.py

#
#  songyulong
#  2022-11-03
#  GNSS 模块库
#  GNSS（Global Navigation Satellite System） 的全称是全球导航卫星系统


# 保存配置参数命令 CAS00
CMD_SAVE_CONFIG = const(b'$PCAS00*01\r\n')  # save configuration information to flash

# 串口通信波特率设置命令 CAS01
CMD_SET_BAUDRATE_4800bps  = const(b'$PCAS01,0*1C\r\n')
CMD_SET_BAUDRATE_9600bps  = const(b'$PCAS01,1*1D\r\n')
CMD_SET_BAUDRATE_19200bps = const(b'$PCAS01,2*1E\r\n')
CMD_SET_BAUDRATE_38400bps = const(b'$PCAS01,3*1F\r\n')
CMD_SET_BAUDRATE_57600bps = const(b'$PCAS01,4*18\r\n')
CMD_SET_BAUDRATE_115200bps= const(b'$PCAS01,5*19\r\n')

# 设定定位更新率 CAS02
CMD_SET_UPDATE_FREQ_1HZ = const(b'$PCAS02,1000*2E\r\n')
CMD_SET_UPDATE_FREQ_2HZ = const(b'$PCAS02,500*1A\r\n')
CMD_SET_UPDATE_FREQ_4HZ = const(b'$PCAS02,250*18\r\n')
CMD_SET_UPDATE_FREQ_5HZ = const(b'$PCAS02,200*1D\r\n')
CMD_SET_UPDATE_FREQ_10HZ= const(b'$PCAS02,100*1E\r\n')

# 设置要求输出或停止输出的 NMEA 语句。CAS03

# 配置工作系统 CAS04  GPS:美国，BDS：中国， GLONASS：俄罗斯
CMD_SET_SYS_GPS              = const(b'$PCAS04,1*18\r\n') 
CMD_SET_SYS_BDS              = const(b'$PCAS04,2*1B\r\n')  
CMD_SET_SYS_GPS_BDS          = const(b'$PCAS04,3*1A\r\n')  
CMD_SET_SYS_GLONASS          = const(b'$PCAS04,4*1D\r\n')
CMD_SET_SYS_GPS_GLONASS      = const(b'$PCAS04,5*1C\r\n')
CMD_SET_SYS_BDS_GLONASS      = const(b'$PCAS04,6*1F\r\n')
CMD_SET_SYS_GPS_BDS_GLONASS  = const(b'$PCAS04,7*1E\r\n')

# 接收机重启 CAS10
CMD_RESTART_Hot     = const(b'$PCAS10,0*1C\r\n') # 热启动 :不使用初始化信息，备份存储中的所有数据有效
CMD_RESTART_Warm    = const(b'$PCAS10,1*1D\r\n') # 温启动 :不使用初始化信息，清除星历
CMD_RESTART_Cold    = const(b'$PCAS10,2*1E\r\n') # 冷启动 :不使用初始化信息，清除备份存储中除配置外的所有数据。
CMD_RESTART_Factory = const(b'$PCAS10,3*1F\r\n') # 出厂启动 :清除内存所有数据，并将接收机复位至出厂默认配置



class NAVIGATION:
    def __init__(self, uart, mode=CMD_SET_SYS_GPS_BDS, update_freq=CMD_SET_UPDATE_FREQ_1HZ):
        self.uart = uart
        self.mode = mode
        self.update_freq = update_freq
        self.longitude = 0
        self.latitude = 0
        self.speed = 0
        self.course = 0
        self.year = 0
        self.month = 0
        self.day = 0
        self.hour = 0
        self.minute = 0
        self.second = 0
        self.rxbuf = bytes()
        self.send_command(self.mode)
        self.send_command(self.update_freq)
        
    def deinit(self):
        self.uart.deinit()
        
    def send_command(self, cmd):
        self.uart.write(cmd)
        
    def recv_loop(self):
        while self.uart.any() > 0:
            #self.rxbuf=self.uart.readline()
            #self.parse_message(self.rxbuf)            
            c = self.uart.read(1)
            self.rxbuf += c
            if c == b'\n': # 一行报文结收结束
                self.parse_message(self.rxbuf)
                self.rxbuf = b''
    
    def parse_message(self,msg):
        try:
            root = msg.decode('utf-8')
            #print(root)
            if (root.startswith('$GNRMC') or root.startswith('$BDRMC') or root.startswith('$GPRMC')): #                 
                GNRMC = root.split(',')
                if GNRMC[2] == 'A':  # 'A':有效, 'V':无效
                    #print(GNRMC)
                    self.hour = int(GNRMC[1][0:2])
                    self.minute = int(GNRMC[1][2:4])
                    self.second = int(GNRMC[1][4:6])
                    #ms = int(GNRMC[1][7:10])
                    
                    self.day = int(GNRMC[9][0:2])
                    self.month = int(GNRMC[9][2:4])
                    self.year = int(GNRMC[9][4:6])

                    if GNRMC[4] == 'N': # 北纬
                        self.latitude = float(GNRMC[3])                        
                    elif GNRMC[4] == 'S': # 南纬
                        self.latitude = -(float(GNRMC[3]))
                        
                    if GNRMC[6] == 'E': # 东经
                        self.longitude = float(GNRMC[5])
                    elif GNRMC[6] == 'W': # 西经
                        self.longitude = -(float(GNRMC[5]))
                        
                    self.speed = float(GNRMC[7])
                    self.course = float(GNRMC[8])
                    #print(self.longitude, self.latitude, self.speed, self.course)
                else:
                    self.latitude = 0
                    self.longitude = 0
                
            elif (root.startswitch('$BDGSV')):
                print('BDGSV')
        except:
            pass
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

from machine import Pin,I2C,UART
from ssd1306 import SSD1306_I2C,SSD1306_SPI
import framebuf

import utime
import navigation
from navigation import *
import _thread

i2c0 = I2C(0, sda=Pin(8), scl=Pin(9),freq=400000)
WIDTH = 128
HEIGHT = 64
oled = SSD1306_I2C(WIDTH,HEIGHT,i2c0)
oled.fill(0)

uart0 = UART(1, baudrate=9600, bits=8, parity=None, stop=1, tx=Pin(4), rx=Pin(5), )
nav = navigation.NAVIGATION(uart0)
nav.send_command(CMD_SET_SYS_BDS)
nav.send_command(CMD_SET_UPDATE_FREQ_1HZ)

# 导航模块结束数据任务
def navigation_task(delay):
    while True:
        nav.recv_loop()
        utime.sleep_ms(delay)
        
_thread.start_new_thread(navigation_task, (1,) )
utime.sleep(1.5)

while True:
    #nav.recv_loop()
    oled.fill(0)
    oled.text("Lat:{:.4f}".format(nav.latitude), 0, 0)
    oled.text("Lon:{:.4f}".format(nav.longitude), 0, 10)
    oled.text("Speed:{:.2f}".format(nav.speed), 0, 20)
    oled.text("Course:{:.2f}".format(nav.course), 0, 30)
    oled.text("{:02d}/{:02d}/{:02d}".format(nav.year, nav.month, nav.day), 0, 40)
    oled.text("{:02d}:{:02d}:{:02d}".format(nav.hour+8, nav.minute, nav.second), 0, 50)
    oled.show()
    #print("main loop....", nav.longitude, nav.latitude, nav.speed, nav.course)
    utime.sleep(1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

无线串口模块(LoRa 433M)

简单数据透传Demo

from machine import Pin,UART
import utime

class WIRELESS_UART:
    def __init__(self, uart, m0, m1, aux=-1, mode=0):
        self.uart = uart
        self.mode = mode
        self.m0 = Pin(m0, Pin.OUT)
        self.m1 = Pin(m1, Pin.OUT)
        if aux >= 0:
            self.aux = Pin(aux, Pin.IN)
        self.set_mode(mode)
        self.rxbuf = bytes()
        
    def set_mode(self, mode):
        self.mode = mode
        if self.mode == 0:
            self.m0.value(0)
            self.m1.value(0)
        elif self.mode == 1:
            self.m0.value(1)
            self.m1.value(0)
        elif self.mode == 2:
            self.m0.value(0)
            self.m1.value(1)
        elif self.mode == 3:
            self.m0.value(1)
            self.m1.value(1)
        else:
            print('Error mode must be in 0 ~ 3')
            
    def write(self, data):
        self.uart.write(data)
        
    def recv_loop(self):
        while self.uart.any() > 0:
            #self.rxbuf=self.uart.readline()
            #self.parse_message(self.rxbuf)            
            c = self.uart.read(1)
            self.rxbuf += c
            if c == b'\n': # 一行报文结收结束
                print(self.rxbuf)
                self.uart.write(b'\n'+self.rxbuf)
                self.rxbuf = b''    
        
uart0 = UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=Pin(16), rx=Pin(17) )    
wuart = WIRELESS_UART(uart=uart0, m0=19, m1=18, aux=-1, mode=0)        
wuart.set_mode(0) # mode:0,1,2,3

while True:
    #wuart.write(b'good')
    wuart.recv_loop()
    utime.sleep_ms(10)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

支持参数配置Demo

from machine import Pin,UART
import utime
import struct


CMD_READ_CONFIG = bytes([0xC1,0xC1,0xC1]) # 读取模块配置参数，比如：C0 00 00 18 50 44。
CMD_READ_INFO = bytes([0xC3,0xC3,0xC3]) # 读取模块版本信息，比如：C3 30 xx yy；30表示E30系列，
CMD_RESET = bytes([0xC4,0xC4,0xC4]) # 复位模块

# 串口校验模式
UART_8N1 = 0x00<<6
UART_8O1 = 0x01<<6
UART_8E1 = 0x02<<6

# 串口波特率
BAUDRATE_1200   = 0x00<<3
BAUDRATE_2400   = 0x01<<3
BAUDRATE_4800   = 0x02<<3
BAUDRATE_9600   = 0x03<<3
BAUDRATE_19200  = 0x04<<3
BAUDRATE_38400  = 0x05<<3
BAUDRATE_57600  = 0x06<<3
BAUDRATE_115200 = 0x07<<3

# 无线空中速率
AIRRATE_1KBPS = 0x00
AIRRATE_2KBPS = 0x01
AIRRATE_5KBPS = 0x02
AIRRATE_8KBPS = 0x03
AIRRATE_10KBPS = 0x04
AIRRATE_15KBPS = 0x05
AIRRATE_20KBPS = 0x06
AIRRATE_25KBPS = 0x07


# 定点发生使能
TRS_MODE = 0x00<<7
FIX_MODE = 0x01<<7

# IO驱动
IO_PPO = 0x01<<6 # Push-Pull Output 推挽输出
IO_ODO = 0x00<<6 # Open Drain Output 开漏输出

WAKEUP_250MS  = 0x00<<3
WAKEUP_500MS  = 0x01<<3
WAKEUP_750MS  = 0x02<<3
WAKEUP_1000MS = 0x03<<3
WAKEUP_1250MS = 0x04<<3
WAKEUP_1500MS = 0x05<<3
WAKEUP_1750MS = 0x06<<3
WAKEUP_2000MS = 0x07<<3

# FEC开关
FEC_ON = 0x01<<2
FEC_OFF = 0x00<<2

# 发射功率
POWER_10dBm = 0x00
POWER_9dBm  = 0x01
POWER_8dBm  = 0x02
POWER_7dBm  = 0x03



class WIRELESS_UART:
    def __init__(self, uart, m0, m1, aux=-1, mode=0):
        self.uart = uart
        self.mode = mode
        self.m0 = Pin(m0, Pin.OUT)
        self.m1 = Pin(m1, Pin.OUT)
        if aux >= 0:
            self.aux = Pin(aux, Pin.IN)
        self.HEAD = 0xC2 # 0xC0:设置参数到ROM， 0xC2:设置参数到RAM
        self.ADDH = 0x00 # 地址高字节 00H-FFH
        self.ADDL = 0x00 # 地址低字节 00H-FFH
        self.SPED = UART_8N1|BAUDRATE_9600|AIRRATE_1KBPS # 速率参数
        self.CHAN = 0xFF # 通信频率 信道
        self.OPTION = TRS_MODE|IO_PPO|WAKEUP_250MS|FEC_ON|POWER_10dBm  #选项
        print(self.SPED, self.OPTION)
        self.set_mode(mode)
        self.rxbuf = bytes()
        self.recv = bytes()
        
    def send_configurate(self, save=False):
        mode = self.mode
        self.set_mode(3)
        utime.sleep_ms(100)
        if save==True:
            self.HEAD = 0xC0
        else:
            self.HEAD = 0xC2
        cmd = bytearray([self.HEAD, self.ADDH, self.ADDL, self.SPED, self.CHAN, self.OPTION])
        print(cmd)
        self.uart.write(cmd)
        utime.sleep_ms(200)
        self.set_mode(mode)
        
    def set_mode(self, mode):
        self.mode = mode
        # mode_0: 一般模式
        # mode_1: 唤醒模式
        # mode_2: 省电模式
        # mode_3: 休眠模式
        if self.mode == 0:
            self.m0.value(0)
            self.m1.value(0)
        elif self.mode == 1:
            self.m0.value(1)
            self.m1.value(0)
        elif self.mode == 2:
            self.m0.value(0)
            self.m1.value(1)
        elif self.mode == 3:
            self.m0.value(1)
            self.m1.value(1)
        else:
            print('Error mode must be in 0 ~ 3')
    
    def read_configurate(self):
        mode = self.mode
        self.set_mode(3)
        self.uart.write(CMD_READ_CONFIG)
        utime.sleep_ms(200)
        recv = bytes()
        while self.uart.any():
            recv +=  self.uart.read(1)
        self.set_mode(mode)
        print(len(recv))
        if len(recv)>=6:
            umode = (recv[3]&0xC0)>>6 # [7,6] 串口校验位模式
            ubsp = (recv[3]&0x38)>>3 # [5,4,3] TTL串口速率(bps)
            wbps = (recv[3]&0x07)    # [2,1,0] 无线空中速率(bps)
            freq = 425+(int(recv[4]) * 0.1) # 通信频率 425M+CHAN*0.1M  00H-FFH,对应425~455MHz
            fcc = (recv[5]&0x04)>>2 #[2] FCC开关 0：关闭FCC, 1:打开FCC
            dBm = (recv[5]&0x03) #[1,0] 发射功率
            print('----------Configurate-------------')
            print('-- head {:#X}'.format(recv[0]))
            print('-- ADDH {:#X}'.format(recv[1]))
            print('-- ADDL {:#X}'.format(recv[2]))
            print('-- SPEED:{:x}'.format(umode))
            print('-- ubsp:{:x}'.format(ubsp))
            print('-- wbps:{:x}'.format(wbps))
            print('-- ch:{:x}'.format(recv[4]))
            print('-- freq:{:d}MHz'.format(freq))
            print('-- option:{:x}'.format(recv[5]))
            print('-- FCC:{:d}'.format(fcc))
            print('-- dBm:{:d}dBm'.format(dBm))
            
        else:
            print('error configuarte head')
        
        return recv
    
    def write(self, data):
        self.uart.write(data)
        
    def recv_loop(self):
        while self.uart.any() > 0:
            #self.rxbuf=self.uart.readline()
            #self.parse_message(self.rxbuf)            
            c = self.uart.read(1)
            self.rxbuf += c
            #print(c)
            if c == b'\n': # 一行报文结收结束
                print(self.rxbuf)
                self.uart.write(b'\n'+self.rxbuf)
                self.rxbuf = b''    
        
uart0 = UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=Pin(16), rx=Pin(17) )    
wuart = WIRELESS_UART(uart=uart0, m0=19, m1=18, aux=-1, mode=0)        

wuart.set_mode(0)

wuart.send_configurate(False)
utime.sleep(1)
ret = wuart.read_configurate()
for r in ret:
    print(r)

utime.sleep(1)
while True:
    #wuart.write(b'good')
    wuart.recv_loop()
    utime.sleep_ms(100)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

旋转编码器

from machine import Pin
import utime

key = Pin(18, Pin.IN)
rotary_clk = Pin(19, Pin.IN)
rotary_dt  = Pin(20, Pin.IN)

def rotary_encoder_task(pin):
    # 当CLK下降沿时，DT是低电平：顺时针旋转，DT是高电平：逆时针旋转
    if rotary_dt.value() == 0:
        print('----right---')
    else:
        print('----left---')
    
rotary_clk.irq(trigger=Pin.IRQ_FALLING, handler=rotary_encoder_task)

# 当按键按下或抬起时会有抖动，这里增加个变量以便过滤掉干扰脉冲
key_pressed = False
def key_task(pin):
    global key_pressed
    if key_pressed == False:
        if key.value()==0:
            key_pressed = True
            print('key down')
    elif key_pressed == True:
        if key.value()==1:
            key_pressed = False
            print('key up')
        
key.irq(trigger=Pin.IRQ_FALLING|Pin.IRQ_RISING, handler=key_task)

while True:
    utime.sleep(0.5)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

步进电机

2相4线步进电机

from machine import Pin,PWM
import machine
import utime

delay_ms = 2

# 四个引脚各一根控制线：A~H表示各线时序
#         A     B     C     D     E     F     G     H
# A+      1     1     0     0     0     0     0     1
# A-      0     0     0     1     1     1     0     0
# B+      0     1     1     1     0     0     0     0
# B-      0     0     0     0     0     1     1     1


AP = Pin(32, Pin.OUT)
BP = Pin(33, Pin.OUT)
AN = Pin(25, Pin.OUT)
BN = Pin(26, Pin.OUT)

AP.value(0)
BP.value(0)
AN.value(0)
BN.value(0)

def motor_step(step=0, delay_ms=1):
    if step == 0:
        # A S1: 1 0 0 0
        AP.value(1)
        AN.value(0)
        BP.value(0)
        BN.value(0)    
    elif step == 1:
        # B S2: 1 0 1 0
        AP.value(1)
        AN.value(0)
        BP.value(1)
        BN.value(0)
    elif step == 2:
        # C S3: 0 0 1 0
        AP.value(0)
        AN.value(0)
        BP.value(1)
        BN.value(0)
    elif step == 3:
        # D S4: 0 1 1 0
        AP.value(0)
        AN.value(1)
        BP.value(1)
        BN.value(0)
    elif step == 4:
        # E S5: 0 1 0 0
        AP.value(0)
        AN.value(1)
        BP.value(0)
        BN.value(0)
    elif step == 5:
        # F S6: 0 1 0 1
        AP.value(0)
        AN.value(1)
        BP.value(0)
        BN.value(1)
    elif step == 6:
        # G S7: 0 0 0 1
        AP.value(0)
        AN.value(0)
        BP.value(0)
        BN.value(1)
    elif step == 7:
        # H S8: 1 0 0 1
        AP.value(1)
        AN.value(0)
        BP.value(0)
        BN.value(1)
    utime.sleep_ms(delay_ms)

def motor_control(dir=0, speed=1):
    if dir == 0:
        # 逆时针 旋转 step 7 ~ 0
        for i in range(0, 7):
            motor_step(7-i, speed)
    else:
        # 逆时针 旋转 step 0 ~ 7
        for i in range(0, 7):
            motor_step(i, speed)
    
    
while True:
    #motor_control(0, 2) # 逆时针
    motor_control(1, 2) # 顺时针
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

4相5线步进电机

from machine import Pin,PWM
import machine
import utime

class step_motor():
    '''
    引脚线序：A~H时序，4相5线共阳接法，8拍控制时序
    +-------------------------------------------------------+    
    |      |  A  |  B  |  C  |  D  |  E  |  F  |  G  |  H   |
    +-------------------------------------------------------|
    |COM      +     +     +     +     +     +     +     +   |
    |  A      0     0     1     1     1     1     1     0   |
    |  B      1     0     0     0     1     1     1     1   |
    |  C      1     1     1     0     0     0     1     1   |
    |  D      1     1     1     1     1     0     0     0   |
    +-------------------------------------------------------+
    '''
    def __init__(self, a, b, c, d):
        self._a = Pin(a, Pin.OUT)
        self._b = Pin(b, Pin.OUT)
        self._c = Pin(c, Pin.OUT)
        self._d = Pin(d, Pin.OUT)
        
    def one_step(self, step=0, delay_ms=1):
        if step == 0:
            # A S1: 0 1 1 1
            self._a.value(0)
            self._b.value(1)
            self._c.value(1)
            self._d.value(1)    
        elif step == 1:
            # B S2: 0 0 1 1
            self._a.value(0)
            self._b.value(0)
            self._c.value(1)
            self._d.value(1)  
        elif step == 2:
            # C S3: 1 0 1 1
            self._a.value(1)
            self._b.value(0)
            self._c.value(1)
            self._d.value(1)  
        elif step == 3:
            # D S4: 1 0 0 1
            self._a.value(1)
            self._b.value(0)
            self._c.value(0)
            self._d.value(1)  
        elif step == 4:
            # E S5: 1 1 0 1
            self._a.value(1)
            self._b.value(1)
            self._c.value(0)
            self._d.value(1)  
        elif step == 5:
            # F S6: 1 1 0 0
            self._a.value(1)
            self._b.value(1)
            self._c.value(0)
            self._d.value(0)
        elif step == 6:
            # G S7: 1 1 1 0
            self._a.value(1)
            self._b.value(1)
            self._c.value(1)
            self._d.value(0)  
        elif step == 7:
            # H S8: 0 1 1 0
            self._a.value(0)
            self._b.value(1)
            self._c.value(1)
            self._d.value(0)  
        utime.sleep_ms(delay_ms)
        
    def motor_control(self, dir=0, speed=1):
        if dir == 0:
            # 逆时针 旋转 step 0 ~ 7
            for i in range(0, 7):
                self.one_step(i, speed)            
        else:
            # 逆时针 旋转 step 7 ~ 0
            for i in range(0, 7):
                self.one_step(7-i, speed)
                                
    def go_steps(self, dir=0, steps=0, speed=1):
        for i in range(steps):
            if dir == 0:
                self.one_step(7-(i%8), speed)
            else:
                self.one_step((i%8), speed)
        
        
        
motor = step_motor(0, 2, 15, 13)

# 此电机转速比 1:64
# 步距角度：5.625/64=0.087890625度
# 360度/0.087890625度=4096个脉冲
# motor_control执行一次产生8个脉冲，4096/8=512

# motor_control执行512次，电机输出刚好旋转一圈360角度
for i in range(512*5): #
    motor.motor_control(1, 2)


# 走4096步，电机输出转一圈
motor.go_steps(1, 4096, 2)

while True:
    utime.sleep(1)
#    motor.motor_control(0, 2)
#    motor.motor_control(1, 1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

INA219电流/电源监测模块

INA219 是一款具备 I2C 或 SMBUS 兼容接口的分流器和功率监测计。该器件监测分流器电压降和总线电源电压，转换次数和滤波选项可通过编程设定。可编程校准值与内部乘法器相结合，支持直接读取电流值（单位：安培）。通过附加乘法寄存器可计算功率（单位：瓦）。I2C 或 SMBUS 兼容接口 具有 16 个可编程地址。INA219 可在 0V 至 26V 范围内感测总线中的分压。该器件由 3V 至 5.5V 单电源供电，电源的最大流耗为1mA。INA219 的工作温度范围为 -40°C 至 125°C。

特性：

• 监测电压范围：0V 至 26V
• 可监测电流、电压、和功率
• 16 个可编程地址
• 高精度：整个温度范围内的精度为 0.5%（最大
  值）(INA219B)
• 滤波选项
• 校准寄存器
  

Demo

from machine import Pin,UART,ADC,I2C
import machine
import utime
import _thread
import struct
import math
import ws2812b

from ssd1306 import SSD1306_I2C,SSD1306_SPI
import framebuf
import array
import utime

from ina219 import INA219,DeviceRangeError


i2c0 = I2C(0, sda=Pin(0), scl=Pin(1),freq=400000)
devlist = i2c0.scan()
print("I2C0 Address List:",devlist)
for dev in devlist:
    print(hex(dev))

i2c1 = I2C(1, sda=Pin(6), scl=Pin(7),freq=400000)
devlist = i2c1.scan()
print("I2C1 Address List:",devlist)
for dev in devlist:
    print(hex(dev))
    
# Raspberry Pi logo as 32x32 bytearray
buffer = bytearray(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00|?\x00\x01\x86@\x80\x01\x01\x80\x80\x01\x11\x88\x80\x01\x05\xa0\x80\x00\x83\xc1\x00\x00C\xe3\x00\x00~\xfc\x00\x00L'\x00\x00\x9c\x11\x00\x00\xbf\xfd\x00\x00\xe1\x87\x00\x01\xc1\x83\x80\x02A\x82@\x02A\x82@\x02\xc1\xc2@\x02\xf6>\xc0\x01\xfc=\x80\x01\x18\x18\x80\x01\x88\x10\x80\x00\x8c!\x00\x00\x87\xf1\x00\x00\x7f\xf6\x00\x008\x1c\x00\x00\x0c \x00\x00\x03\xc0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")

# Load the raspberry pi logo into the framebuffer (the image is 32x32)
fb = framebuf.FrameBuffer(buffer, 32, 32, framebuf.MONO_HLSB)

WIDTH = 128
HEIGHT = 64

try:
    oled = SSD1306_I2C(WIDTH,HEIGHT,i2c0)    
except:
    pass

oled.fill(0)
# oled.blit(fb, (128-32)//2, (64-32)//2)
#oled.text('RaspberryPi Pico', 0, 0)
#oled.hline(0,12, 128, 1)
#oled.blit(fb, (128-32), 15)
#oled.show()


# 分路电阻值，单位为欧姆（强制）
SHUNT_OHMS = 0.1

# 以安培为单位的最大预期电流（可选）
MAX_EXPECTED_AMPS = 3 #1
 
# 电阻值，通讯协议，电流值
ina = INA219(SHUNT_OHMS, i2c0, MAX_EXPECTED_AMPS)
 
# 配置和校准INA219的测量方式OUT输出12位
#ina.configure(ina.RANGE_32V)
ina.configure(voltage_range=ina.RANGE_32V, gain=ina.GAIN_AUTO,
                  bus_adc=ina.ADC_16SAMP, shunt_adc=ina.ADC_16SAMP)

while True:
    voltage = ina.voltage()
    current = ina.current()
    power = ina.power()
    shunt = ina.shunt_voltage()
    
    print("电压:", voltage, "V")
    print("电流:", current, "mA")
    print("功率:", power, "mW")
    print("分压:", shunt, "mV")
    
    oled.fill(0)
    oled.text("Voltage:{:.2f}V".format(voltage), 0, 0)
    oled.text("Current:{:.2f}mA".format(current), 0, 12)
    oled.text("Power:{:.2f}mW".format(power), 0, 12*2)
    oled.text("shunt:{:.2f}mV".format(shunt), 0, 12*3)
    oled.show()
    utime.sleep(0.2)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

驱动库文件：ina219.py

"""MicroPython library for the INA219 sensor.
This library supports the INA219 sensor from Texas Instruments with
MicroPython using the I2C bus.
"""

import utime
from math import trunc
from micropython import const
 
 
class INA219:
    """Provides all the functionality to interact with the INA219 sensor."""
 
    RANGE_16V = const(0)  # Range 0-16 volts
    RANGE_32V = const(1)  # Range 0-32 volts
 
    GAIN_1_40MV = const(0)  # Maximum shunt voltage 40mV
    GAIN_2_80MV = const(1)  # Maximum shunt voltage 80mV
    GAIN_4_160MV = const(2)  # Maximum shunt voltage 160mV
    GAIN_8_320MV = const(3)  # Maximum shunt voltage 320mV
    GAIN_AUTO = const(-1)  # Determine gain automatically
 
    ADC_9BIT = const(0)  # 9-bit conversion time  84us.
    ADC_10BIT = const(1)  # 10-bit conversion time 148us.
    ADC_11BIT = const(2)  # 11-bit conversion time 2766us.
    ADC_12BIT = const(3)  # 12-bit conversion time 532us.
    ADC_2SAMP = const(9)  # 2 samples at 12-bit, conversion time 1.06ms.
    ADC_4SAMP = const(10)  # 4 samples at 12-bit, conversion time 2.13ms.
    ADC_8SAMP = const(11)  # 8 samples at 12-bit, conversion time 4.26ms.
    ADC_16SAMP = const(12)  # 16 samples at 12-bit,conversion time 8.51ms
    ADC_32SAMP = const(13)  # 32 samples at 12-bit, conversion time 17.02ms.
    ADC_64SAMP = const(14)  # 64 samples at 12-bit, conversion time 34.05ms.
    ADC_128SAMP = const(15)  # 128 samples at 12-bit, conversion time 68.10ms.
 
    __ADC_CONVERSION = {
        ADC_9BIT: "9-bit",
        ADC_10BIT: "10-bit",
        ADC_11BIT: "11-bit",
        ADC_12BIT: "12-bit",
        ADC_2SAMP: "12-bit, 2 samples",
        ADC_4SAMP: "12-bit, 4 samples",
        ADC_8SAMP: "12-bit, 8 samples",
        ADC_16SAMP: "12-bit, 16 samples",
        ADC_32SAMP: "12-bit, 32 samples",
        ADC_64SAMP: "12-bit, 64 samples",
        ADC_128SAMP: "12-bit, 128 samples"
    }
 
    __ADDRESS = 0x40
 
    __REG_CONFIG = 0x00
    __REG_SHUNTVOLTAGE = 0x01
    __REG_BUSVOLTAGE = 0x02
    __REG_POWER = 0x03
    __REG_CURRENT = 0x04
    __REG_CALIBRATION = 0x05
 
    __RST = 15
    __BRNG = 13
    __PG1 = 12
    __PG0 = 11
    __BADC4 = 10
    __BADC3 = 9
    __BADC2 = 8
    __BADC1 = 7
    __SADC4 = 6
    __SADC3 = 5
    __SADC2 = 4
    __SADC1 = 3
    __MODE3 = 2
    __MODE2 = 1
    __MODE1 = 0
 
    __OVF = 1
    __CNVR = 2
 
    __BUS_RANGE = [16, 32]
    __GAIN_VOLTS = [0.04, 0.08, 0.16, 0.32]
 
    __CONT_SH_BUS = 7
 
    __AMP_ERR_MSG = ('Expected current %.3fA is greater '
                     'than max possible current %.3fA')
    __RNG_ERR_MSG = ('Expected amps %.2fA, out of range, use a lower '
                     'value shunt resistor')
    __VOLT_ERR_MSG = ('Invalid voltage range, must be one of: '
                      'RANGE_16V, RANGE_32V')
 
    __LOG_FORMAT = '%(asctime)s - %(levelname)s - INA219 %(message)s'
    __LOG_MSG_1 = ('shunt ohms: %.3f, bus max volts: %d, '
                   'shunt volts max: %.2f%s, '
                   'bus ADC: %s, shunt ADC: %s')
    __LOG_MSG_2 = ('calibrate called with: bus max volts: %dV, '
                   'max shunt volts: %.2fV%s')
    __LOG_MSG_3 = ('Current overflow detected - '
                   'attempting to increase gain')
 
    __SHUNT_MILLIVOLTS_LSB = 0.01  # 10uV
    __BUS_MILLIVOLTS_LSB = 4  # 4mV
    __CALIBRATION_FACTOR = 0.04096
    # Max value supported value (65534 decimal) of the calibration register
    # (D0 bit is always zero, p31 of spec)
    __MAX_CALIBRATION_VALUE = 0xFFFE
    # In the spec (p17) the current LSB factor for the minimum LSB is
    # documented as 32767, but a larger value (100.1% of 32767) is used
    # to guarantee that current overflow can always be detected.
    __CURRENT_LSB_FACTOR = 32800
 
    def __init__(self, shunt_ohms, i2c, max_expected_amps=None,
                 address=__ADDRESS):
        """Construct the class.
        At a minimum pass in the resistance of the shunt resistor and I2C
        interface to which the sensor is connected.
        Arguments:
        shunt_ohms -- value of shunt resistor in Ohms (mandatory).
        i2c -- an instance of the I2C class from the *machine* module, either
            I2C(1) or I2C(2) (mandatory).
        max_expected_amps -- the maximum expected current in Amps (optional).
        address -- the I2C address of the INA219, defaults to
            *0x40* (optional).
        log_level -- set to logging.DEBUG to see detailed calibration
            calculations (optional).
        """
       
        
        self._i2c = i2c
        self._address = address
        self._shunt_ohms = shunt_ohms
        self._max_expected_amps = max_expected_amps
        self._min_device_current_lsb = self._calculate_min_current_lsb()
        self._gain = None
        self._auto_gain_enabled = False
 
    def configure(self, voltage_range=RANGE_32V, gain=GAIN_AUTO,
                  bus_adc=ADC_12BIT, shunt_adc=ADC_12BIT):
        """Configure and calibrate how the INA219 will take measurements.
        Arguments:
        voltage_range -- The full scale voltage range, this is either 16V
            or 32V represented by one of the following constants;
            RANGE_16V, RANGE_32V (default).
        gain -- The gain which controls the maximum range of the shunt
            voltage represented by one of the following constants;
            GAIN_1_40MV, GAIN_2_80MV, GAIN_4_160MV,
            GAIN_8_320MV, GAIN_AUTO (default).
        bus_adc -- The bus ADC resolution (9, 10, 11, or 12-bit) or
            set the number of samples used when averaging results
            represent by one of the following constants; ADC_9BIT,
            ADC_10BIT, ADC_11BIT, ADC_12BIT (default),
            ADC_2SAMP, ADC_4SAMP, ADC_8SAMP, ADC_16SAMP,
            ADC_32SAMP, ADC_64SAMP, ADC_128SAMP
        shunt_adc -- The shunt ADC resolution (9, 10, 11, or 12-bit) or
            set the number of samples used when averaging results
            represent by one of the following constants; ADC_9BIT,
            ADC_10BIT, ADC_11BIT, ADC_12BIT (default),
            ADC_2SAMP, ADC_4SAMP, ADC_8SAMP, ADC_16SAMP,
            ADC_32SAMP, ADC_64SAMP, ADC_128SAMP
        """
        self.__validate_voltage_range(voltage_range)
        self._voltage_range = voltage_range
 
        if self._max_expected_amps is not None:
            if gain == self.GAIN_AUTO:
                self._auto_gain_enabled = True
                self._gain = self._determine_gain(self._max_expected_amps)
            else:
                self._gain = gain
        else:
            if gain != self.GAIN_AUTO:
                self._gain = gain
            else:
                self._auto_gain_enabled = True
                self._gain = self.GAIN_1_40MV
 
        
 
        self._calibrate(
            self.__BUS_RANGE[voltage_range], self.__GAIN_VOLTS[self._gain],
            self._max_expected_amps)
        self._configure(voltage_range, self._gain, bus_adc, shunt_adc)
 
    def voltage(self):
        """Return the bus voltage in volts."""
        value = self._voltage_register()
        return float(value) * self.__BUS_MILLIVOLTS_LSB / 1000
 
    def supply_voltage(self):
        """Return the bus supply voltage in volts.
        This is the sum of the bus voltage and shunt voltage. A
        DeviceRangeError exception is thrown if current overflow occurs.
        """
        return self.voltage() + (float(self.shunt_voltage()) / 1000)
 
    def current(self):
        """Return the bus current in milliamps.
        A DeviceRangeError exception is thrown if current overflow occurs.
        """
        self._handle_current_overflow()
        return self._current_register() * self._current_lsb * 1000
 
    def power(self):
        """Return the bus power consumption in milliwatts.
        A DeviceRangeError exception is thrown if current overflow occurs.
        """
        self._handle_current_overflow()
        return self._power_register() * self._power_lsb * 1000
 
    def shunt_voltage(self):
        """Return the shunt voltage in millivolts.
        A DeviceRangeError exception is thrown if current overflow occurs.
        """
        self._handle_current_overflow()
        return self._shunt_voltage_register() * self.__SHUNT_MILLIVOLTS_LSB
 
    def sleep(self):
        """Put the INA219 into power down mode."""
        configuration = self._read_configuration()
        self._configuration_register(configuration & 0xFFF8)
 
    def wake(self):
        """Wake the INA219 from power down mode."""
        configuration = self._read_configuration()
        self._configuration_register(configuration | 0x0007)
        # 40us delay to recover from powerdown (p14 of spec)
        utime.sleep_us(40)
 
    def current_overflow(self):
        """Return true if the sensor has detect current overflow.
        In this case the current and power values are invalid.
        """
        return self._has_current_overflow()
 
    def reset(self):
        """Reset the INA219 to its default configuration."""
        self._configuration_register(1 << self.__RST)
 
    def _handle_current_overflow(self):
        if self._auto_gain_enabled:
            while self._has_current_overflow():
                self._increase_gain()
        else:
            if self._has_current_overflow():
                raise DeviceRangeError(self.__GAIN_VOLTS[self._gain])
 
    def _determine_gain(self, max_expected_amps):
        shunt_v = max_expected_amps * self._shunt_ohms
        if shunt_v > self.__GAIN_VOLTS[3]:
            raise ValueError(self.__RNG_ERR_MSG % max_expected_amps)
        gain = min(v for v in self.__GAIN_VOLTS if v > shunt_v)
        return self.__GAIN_VOLTS.index(gain)
 
    def _increase_gain(self):        
        gain = self._read_gain()
        if gain < len(self.__GAIN_VOLTS) - 1:
            gain = gain + 1
            self._calibrate(self.__BUS_RANGE[self._voltage_range],
                            self.__GAIN_VOLTS[gain])
            self._configure_gain(gain)
            # 1ms delay required for new configuration to take effect,
            # otherwise invalid current/power readings can occur.
            utime.sleep_ms(1)
        else:           
            raise DeviceRangeError(self.__GAIN_VOLTS[gain], True)
 
    def _configure(self, voltage_range, gain, bus_adc, shunt_adc):
        configuration = (
            voltage_range << self.__BRNG | gain << self.__PG0 |
            bus_adc << self.__BADC1 | shunt_adc << self.__SADC1 |
            self.__CONT_SH_BUS)
        self._configuration_register(configuration)
 
    def _calibrate(self, bus_volts_max, shunt_volts_max,
                   max_expected_amps=None):        
 
        max_possible_amps = shunt_volts_max / self._shunt_ohms
        
 
        self._current_lsb = \
            self._determine_current_lsb(max_expected_amps, max_possible_amps)
        self._power_lsb = self._current_lsb * 20        
    
        max_current = self._current_lsb * 32767
    
        max_shunt_voltage = max_current * self._shunt_ohms
        
 
        calibration = trunc(self.__CALIBRATION_FACTOR /
                            (self._current_lsb * self._shunt_ohms))        
        self._calibration_register(calibration)
 
    def _determine_current_lsb(self, max_expected_amps, max_possible_amps):
        if max_expected_amps is not None:
            if max_expected_amps > round(max_possible_amps, 3):
                raise ValueError(self.__AMP_ERR_MSG %
                                 (max_expected_amps, max_possible_amps))
            
            if max_expected_amps < max_possible_amps:
                current_lsb = max_expected_amps / self.__CURRENT_LSB_FACTOR
            else:
                current_lsb = max_possible_amps / self.__CURRENT_LSB_FACTOR
        else:
            current_lsb = max_possible_amps / self.__CURRENT_LSB_FACTOR
 
        if current_lsb < self._min_device_current_lsb:
            current_lsb = self._min_device_current_lsb
        return current_lsb
 
    def _configuration_register(self, register_value):
        
        self.__write_register(self.__REG_CONFIG, register_value)
 
    def _read_configuration(self):
        return self.__read_register(self.__REG_CONFIG)
 
    def _calculate_min_current_lsb(self):
        return self.__CALIBRATION_FACTOR / \
            (self._shunt_ohms * self.__MAX_CALIBRATION_VALUE)
 
    def _read_gain(self):
        configuration = self._read_configuration()
        gain = (configuration & 0x1800) >> self.__PG0
        
        return gain
 
    def _configure_gain(self, gain):
        configuration = self._read_configuration()
        configuration = configuration & 0xE7FF
        self._configuration_register(configuration | (gain << self.__PG0))
        self._gain = gain
        
 
    def _calibration_register(self, register_value):
        self.__write_register(self.__REG_CALIBRATION, register_value)
 
    def _has_current_overflow(self):
        ovf = self._read_voltage_register() & self.__OVF
        return (ovf == 1)
 
    def _voltage_register(self):
        register_value = self._read_voltage_register()
        return register_value >> 3
 
    def _read_voltage_register(self):
        return self.__read_register(self.__REG_BUSVOLTAGE)
 
    def _current_register(self):
        return self.__read_register(self.__REG_CURRENT, True)
 
    def _shunt_voltage_register(self):
        return self.__read_register(self.__REG_SHUNTVOLTAGE, True)
 
    def _power_register(self):
        return self.__read_register(self.__REG_POWER)
 
    def __validate_voltage_range(self, voltage_range):
        if voltage_range > len(self.__BUS_RANGE) - 1:
            raise ValueError(self.__VOLT_ERR_MSG)
 
    def __write_register(self, register, register_value):
        self.__log_register_operation("write", register, register_value)
 
        register_bytes = self.__to_bytes(register_value)
        self._i2c.writeto_mem(self._address, register, register_bytes)
 
    def __to_bytes(self, register_value):
        return bytearray([(register_value >> 8) & 0xFF, register_value & 0xFF])
 
    def __read_register(self, register, negative_value_supported=False):
        register_bytes = self._i2c.readfrom_mem(self._address, register, 2)
        register_value = int.from_bytes(register_bytes, 'big')
        if negative_value_supported:
            # Two's compliment
            if register_value > 32767:
                register_value -= 65536
 
        self.__log_register_operation("read", register, register_value)
        return register_value
 
    def __log_register_operation(self, msg, register, value):
        # performance optimisation
        #if logging._level == logging.DEBUG:
        #    binary = '{0:#018b}'.format(value)
        #    self._log.debug("%s register 0x%02x: 0x%04x %s",
        #                    msg, register, value, binary)
         pass
 
    def __max_expected_amps_to_string(self, max_expected_amps):
        if max_expected_amps is None:
            return ''
        else:
            return ', max expected amps: %.3fA' % max_expected_amps
 
 
class DeviceRangeError(Exception):
    """This exception is throw to prevent invalid readings.
    Invalid readings occur When the current is greater than allowed given
    calibration of the device.
    """
 
    __DEV_RNG_ERR = ('Current out of range (overflow), '
                     'for gain %.2fV')
 
    def __init__(self, gain_volts, device_max=False):
        """Construct the class."""
        msg = self.__DEV_RNG_ERR % gain_volts
        if device_max:
            msg = msg + ', device limit reached'
        super(DeviceRangeError, self).__init__(msg)
        self.gain_volts = gain_volts
        self.device_limit_reached = device_max

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410

模块接线示意图：

PJ7620手势识别传感器

PAJ7620是一款通过红外线感应手势的传感器模块，可以实现识别9种不同的手势动作，包括向上、向下、向左、向右、向前、向后、左旋转、有旋转、波动等。

演示Demo:

from machine import Pin,UART,ADC,I2C
import machine
import utime
import _thread
import struct
import math

from ssd1306 import SSD1306_I2C,SSD1306_SPI
import framebuf
import array

from paj7620 import PAJ7620


led = Pin(25, Pin.OUT)

i2c0 = I2C(0, sda=Pin(0), scl=Pin(1),freq=400000)
devlist = i2c0.scan()
print("I2C0 Address List:",devlist)
for dev in devlist:
    print(hex(dev))
    

 
# Raspberry Pi logo as 32x32 bytearray
buffer = bytearray(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00|?\x00\x01\x86@\x80\x01\x01\x80\x80\x01\x11\x88\x80\x01\x05\xa0\x80\x00\x83\xc1\x00\x00C\xe3\x00\x00~\xfc\x00\x00L'\x00\x00\x9c\x11\x00\x00\xbf\xfd\x00\x00\xe1\x87\x00\x01\xc1\x83\x80\x02A\x82@\x02A\x82@\x02\xc1\xc2@\x02\xf6>\xc0\x01\xfc=\x80\x01\x18\x18\x80\x01\x88\x10\x80\x00\x8c!\x00\x00\x87\xf1\x00\x00\x7f\xf6\x00\x008\x1c\x00\x00\x0c \x00\x00\x03\xc0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
# Load the raspberry pi logo into the framebuffer (the image is 32x32)
fb = framebuf.FrameBuffer(buffer, 32, 32, framebuf.MONO_HLSB)

WIDTH = 128
HEIGHT = 64

try:
    oled = SSD1306_I2C(WIDTH,HEIGHT,i2c0)    
except:
    print("oled initialize error!")
    pass
oled.fill(0)


# 手势传感器
paj = PAJ7620(i2c=i2c0)


def PAJ7260():
    Gesture = [" ","Forward-前","Backward-后","Right-右","Left-左","Up-上","Down-下","Clockwise-顺","anti-clockwise-逆","Wave-波浪"]
        
    print(get)
    if get != 0:
        if Gesture[get] == Gesture[9]:
            led.value(not led.value())
        else:
            led.value(0)
            
def main():
    while True:
        geste = paj.gesture()
        # geste peut contenir les valeurs suivantes
        # 	0 : nothing
        # 	1 : Forward
        # 	2 : Backward
        # 	3 : Right
        # 	4 : Left
        # 	5 : Up
        # 	6 : Down
        # 	7 : Clockwise
        # 	8 : anti-clockwise
        # 	9 : wave
        
        oled.fill(0)  # efface l'ecran
        if geste == 1: 
            oled.text("Forward", 0, 0)
        elif geste == 2:
            oled.text("Backward", 0, 0)
        elif geste == 3:
            oled.text("Right", 0, 0)
        elif geste == 4:
            oled.text("Left", 0, 0)
        elif geste == 5:
            oled.text("Up", 0, 0)
        elif geste == 6:
            oled.text("Down", 0, 0)
        elif geste == 7:
            oled.text("Clockwise", 0, 0)
        elif geste == 8:
            oled.text("anti-clockwise", 0, 0)
        elif geste == 9:
            oled.text("wave", 0, 0)
        else:
            oled.text("Get Gesture", 0, 0)
        oled.show()
        utime.sleep(0.5)

if __name__ == "__main__":
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

驱动库：paj7620.py

#
# Librairie pour l'utilisation du capteur de geste PAJ7620
#
# adaptation pour l'ESP32 du programme Python pour le Raspberry a partir 
# https://github.com/SeeedDocument/wiki_english/blob/master/docs/Grove-Gesture_v1.0.md
#
import machine, time

# Registres et variables pour le capteur de geste
GES_REACTION_TIME = 0.2  # Ajustement du temps de reaction en secondes
GES_ENTRY_TIME = 0.5  # si reconnaissance des gestes Avance/Recule, GES_REACTION_TIME doit etre inferieur a GES_ENTRY_TIME(0.5s).

BANK0 = 0
BANK1 = 1

PAJ7620_ADDR_BASE = 0x00
# registre de selection de la banque
PAJ7620_REGISTER_BANK_SEL = (PAJ7620_ADDR_BASE + 0xEF)	        # W

# adresse i2c
PAJ7620_ID = 0x73

# registres BANK 0
PAJ7620_ADDR_SUSPEND_CMD	    = (PAJ7620_ADDR_BASE + 0x3)	     # W
PAJ7620_ADDR_GES_PS_DET_MASK_0	= (PAJ7620_ADDR_BASE + 0x41)	 # RW
PAJ7620_ADDR_GES_PS_DET_MASK_1	= (PAJ7620_ADDR_BASE + 0x42)	 # RW
PAJ7620_ADDR_GES_PS_DET_FLAG_0	= (PAJ7620_ADDR_BASE + 0x43)	 # R
PAJ7620_ADDR_GES_PS_DET_FLAG_1	= (PAJ7620_ADDR_BASE + 0x44)	 # R
PAJ7620_ADDR_STATE_INDICATOR	= (PAJ7620_ADDR_BASE + 0x45)	 # R
PAJ7620_ADDR_PS_HIGH_THRESHOLD	= (PAJ7620_ADDR_BASE + 0x69)	 # RW
PAJ7620_ADDR_PS_LOW_THRESHOLD	= (PAJ7620_ADDR_BASE + 0x6A)	 # RW
PAJ7620_ADDR_PS_APPROACH_STATE	= (PAJ7620_ADDR_BASE + 0x6B)	 # R
PAJ7620_ADDR_PS_RAW_DATA	    = (PAJ7620_ADDR_BASE + 0x6C)	 # R

# registres BANK 1
PAJ7620_ADDR_PS_GAIN		    = (PAJ7620_ADDR_BASE + 0x44)	# RW
PAJ7620_ADDR_IDLE_S1_STEP_0    = (PAJ7620_ADDR_BASE + 0x67)	    # RW
PAJ7620_ADDR_IDLE_S1_STEP_1    = (PAJ7620_ADDR_BASE + 0x68)	    # RW
PAJ7620_ADDR_IDLE_S2_STEP_0    = (PAJ7620_ADDR_BASE + 0x69)	    # RW
PAJ7620_ADDR_IDLE_S2_STEP_1    = (PAJ7620_ADDR_BASE + 0x6A)	    # RW
PAJ7620_ADDR_OP_TO_S1_STEP_0	= (PAJ7620_ADDR_BASE + 0x6B)	# RW
PAJ7620_ADDR_OP_TO_S1_STEP_1	= (PAJ7620_ADDR_BASE + 0x6C)	# RW
PAJ7620_ADDR_OP_TO_S2_STEP_0	= (PAJ7620_ADDR_BASE + 0x6D)	# RW
PAJ7620_ADDR_OP_TO_S2_STEP_1	= (PAJ7620_ADDR_BASE + 0x6E)	# RW
PAJ7620_ADDR_OPERATION_ENABLE	= (PAJ7620_ADDR_BASE + 0x72)	# RW

# PAJ7620 selection de la banque
PAJ7620_BANK0 = 0
PAJ7620_BANK1 = 1

# gestes detectes
GES_RIGHT_FLAG           = 1<<0
GES_LEFT_FLAG            = 1<<1
GES_UP_FLAG              = 1<<2
GES_DOWN_FLAG            = 1<<3
GES_FORWARD_FLAG         = 1<<4
GES_BACKWARD_FLAG        = 1<<5
GES_CLOCKWISE_FLAG       = 1<<6
GES_COUNT_CLOCKWISE_FLAG = 1<<7
GES_WAVE_FLAG            = 1<<0

# retour des gestes
NOTHING         = 0
FORWARD 		= 1
BACKWARD 		= 2
RIGHT			= 3
LEFT			= 4
UP			    = 5
DOWN			= 6
CLOCKWISE		= 7
ANTI_CLOCKWISE	= 8
WAVE			= 9

# etat initial des differents registres 
initRegisterArray = ([0xEF,0x00],[0x32,0x29],[0x33,0x01],[0x34,0x00],[0x35,0x01],[0x36,0x00],
[0x37,0x07],[0x38,0x17],[0x39,0x06],[0x3A,0x12],[0x3F,0x00],[0x40,0x02],[0x41,0xFF],[0x42,0x01],
[0x46,0x2D],[0x47,0x0F],[0x48,0x3C],[0x49,0x00],[0x4A,0x1E],[0x4B,0x00],[0x4C,0x20],[0x4D,0x00],
[0x4E,0x1A],[0x4F,0x14],[0x50,0x00],[0x51,0x10],[0x52,0x00],[0x5C,0x02],[0x5D,0x00],[0x5E,0x10],
[0x5F,0x3F],[0x60,0x27],[0x61,0x28],[0x62,0x00],[0x63,0x03],[0x64,0xF7],[0x65,0x03],[0x66,0xD9],
[0x67,0x03],[0x68,0x01],[0x69,0xC8],[0x6A,0x40],[0x6D,0x04],[0x6E,0x00],[0x6F,0x00],[0x70,0x80],
[0x71,0x00],[0x72,0x00],[0x73,0x00],[0x74,0xF0],[0x75,0x00],[0x80,0x42],[0x81,0x44],[0x82,0x04],
[0x83,0x20],[0x84,0x20],[0x85,0x00],[0x86,0x10],[0x87,0x00],[0x88,0x05],[0x89,0x18],[0x8A,0x10],
[0x8B,0x01],[0x8C,0x37],[0x8D,0x00],[0x8E,0xF0],[0x8F,0x81],[0x90,0x06],[0x91,0x06],[0x92,0x1E],
[0x93,0x0D],[0x94,0x0A],[0x95,0x0A],[0x96,0x0C],[0x97,0x05],[0x98,0x0A],[0x99,0x41],[0x9A,0x14],
[0x9B,0x0A],[0x9C,0x3F],[0x9D,0x33],[0x9E,0xAE],[0x9F,0xF9],[0xA0,0x48],[0xA1,0x13],[0xA2,0x10],
[0xA3,0x08],[0xA4,0x30],[0xA5,0x19],[0xA6,0x10],[0xA7,0x08],[0xA8,0x24],[0xA9,0x04],[0xAA,0x1E],
[0xAB,0x1E],[0xCC,0x19],[0xCD,0x0B],[0xCE,0x13],[0xCF,0x64],[0xD0,0x21],[0xD1,0x0F],[0xD2,0x88],
[0xE0,0x01],[0xE1,0x04],[0xE2,0x41],[0xE3,0xD6],[0xE4,0x00],[0xE5,0x0C],[0xE6,0x0A],[0xE7,0x00],
[0xE8,0x00],[0xE9,0x00],[0xEE,0x07],[0xEF,0x01],[0x00,0x1E],[0x01,0x1E],[0x02,0x0F],[0x03,0x10],
[0x04,0x02],[0x05,0x00],[0x06,0xB0],[0x07,0x04],[0x08,0x0D],[0x09,0x0E],[0x0A,0x9C],[0x0B,0x04],
[0x0C,0x05],[0x0D,0x0F],[0x0E,0x02],[0x0F,0x12],[0x10,0x02],[0x11,0x02],[0x12,0x00],[0x13,0x01],
[0x14,0x05],[0x15,0x07],[0x16,0x05],[0x17,0x07],[0x18,0x01],[0x19,0x04],[0x1A,0x05],[0x1B,0x0C],
[0x1C,0x2A],[0x1D,0x01],[0x1E,0x00],[0x21,0x00],[0x22,0x00],[0x23,0x00],[0x25,0x01],[0x26,0x00],
[0x27,0x39],[0x28,0x7F],[0x29,0x08],[0x30,0x03],[0x31,0x00],[0x32,0x1A],[0x33,0x1A],[0x34,0x07],
[0x35,0x07],[0x36,0x01],[0x37,0xFF],[0x38,0x36],[0x39,0x07],[0x3A,0x00],[0x3E,0xFF],[0x3F,0x00],
[0x40,0x77],[0x41,0x40],[0x42,0x00],[0x43,0x30],[0x44,0xA0],[0x45,0x5C],[0x46,0x00],[0x47,0x00],
[0x48,0x58],[0x4A,0x1E],[0x4B,0x1E],[0x4C,0x00],[0x4D,0x00],[0x4E,0xA0],[0x4F,0x80],[0x50,0x00],
[0x51,0x00],[0x52,0x00],[0x53,0x00],[0x54,0x00],[0x57,0x80],[0x59,0x10],[0x5A,0x08],[0x5B,0x94],
[0x5C,0xE8],[0x5D,0x08],[0x5E,0x3D],[0x5F,0x99],[0x60,0x45],[0x61,0x40],[0x63,0x2D],[0x64,0x02],
[0x65,0x96],[0x66,0x00],[0x67,0x97],[0x68,0x01],[0x69,0xCD],[0x6A,0x01],[0x6B,0xB0],[0x6C,0x04],
[0x6D,0x2C],[0x6E,0x01],[0x6F,0x32],[0x71,0x00],[0x72,0x01],[0x73,0x35],[0x74,0x00],[0x75,0x33],
[0x76,0x31],[0x77,0x01],[0x7C,0x84],[0x7D,0x03],[0x7E,0x01])

class PAJ7620():
    def __init__(self, i2c, address=PAJ7620_ID):
        self.i2c = i2c
        self.address = address
        time.sleep(0.001)
        self.paj7620SelectBank(BANK0)
        data0 = self.paj7620ReadReg(0, 1)[0]
        data1 = self.paj7620ReadReg(1, 1)[0]
        if data0 != 0x20 or data1 != 0x76:
            print("Capteur en erreur")
        for i in range(len(initRegisterArray)):
            self.paj7620WriteReg(initRegisterArray[i][0], initRegisterArray[i][1])
        self.paj7620SelectBank(BANK0)
        
    def paj7620WriteReg(self, addr, cmd):
        # Ecriture d'un octet dans un registre du capteur
        buf = bytearray(1)
        buf[0] = cmd
        self.i2c.writeto_mem(self.address, addr, buf)
	
    def paj7620SelectBank(self, bank):
        # Selection de la banque de registre
        if bank == BANK0:
            self.paj7620WriteReg(PAJ7620_REGISTER_BANK_SEL, PAJ7620_BANK0)
    
    def paj7620ReadReg(self, addr, qty):
        # Lecture d'un bloc a l'adresse 'addr' avec un nombre d'octets 'qty'
        return self.i2c.readfrom_mem(self.address, addr, qty)
    
    def gesture(self):
        # Retourne une valeur à partir du capteur
        # 	0 : nothing
        # 	1 : Forward
        # 	2 : Backward
        # 	3 : Right
        # 	4 : Left
        # 	5 : Up
        # 	6 : Down
        # 	7 : Clockwise
        # 	8 : anti-clockwise
        # 	9 : wave
        data = self.paj7620ReadReg(0x43, 1)[0]
        if data == GES_RIGHT_FLAG:
            time.sleep(GES_ENTRY_TIME)
            data = self.paj7620ReadReg(0x43, 1)[0]
            if data == GES_FORWARD_FLAG:
                return FORWARD
            elif data == GES_BACKWARD_FLAG:
                return BACKWARD
            else:
                return RIGHT
        elif data == GES_LEFT_FLAG:
            time.sleep(GES_ENTRY_TIME)
            data = self.paj7620ReadReg(0x43, 1)[0]
            if data == GES_FORWARD_FLAG:
                return FORWARD
            elif data == GES_BACKWARD_FLAG:
                return BACKWARD
            else:
                return LEFT
        elif data == GES_UP_FLAG:
            time.sleep(GES_ENTRY_TIME)
            data = self.paj7620ReadReg(0x43, 1)[0]
            if data == GES_FORWARD_FLAG:
                return FORWARD
            elif data == GES_BACKWARD_FLAG:
                return BACKWARD
            else:
                return UP
        elif data == GES_DOWN_FLAG:
            time.sleep(GES_ENTRY_TIME)
            data = self.paj7620ReadReg(0x43, 1)[0]
            if data == GES_FORWARD_FLAG:
                return FORWARD
            elif data == GES_BACKWARD_FLAG:
                return BACKWARD
            else:
                return DOWN
        elif data == GES_FORWARD_FLAG:
            return FORWARD
        elif data == GES_BACKWARD_FLAG:
            return BACKWARD
        elif data == GES_CLOCKWISE_FLAG:
            return CLOCKWISE
        elif data == GES_COUNT_CLOCKWISE_FLAG:
            return ANTI_CLOCKWISE
        else:
            data1 = self.paj7620ReadReg(0x44, 1)[0]
            if (data1 == GES_WAVE_FLAG):
                return WAVE
        return NOTHING

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195

HX711称重模块

演示Demo

from machine import Pin, I2C
import utime
from hx711 import HX711

from ssd1306 import SSD1306_I2C,SSD1306_SPI
import framebuf
import array


i2c0 = I2C(0, sda=Pin(0), scl=Pin(1),freq=400000)
devlist = i2c0.scan()
print("I2C0 Address List:",devlist)
for dev in devlist:
    print(hex(dev))
    
    # Raspberry Pi logo as 32x32 bytearray
buffer = bytearray(b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00|?\x00\x01\x86@\x80\x01\x01\x80\x80\x01\x11\x88\x80\x01\x05\xa0\x80\x00\x83\xc1\x00\x00C\xe3\x00\x00~\xfc\x00\x00L'\x00\x00\x9c\x11\x00\x00\xbf\xfd\x00\x00\xe1\x87\x00\x01\xc1\x83\x80\x02A\x82@\x02A\x82@\x02\xc1\xc2@\x02\xf6>\xc0\x01\xfc=\x80\x01\x18\x18\x80\x01\x88\x10\x80\x00\x8c!\x00\x00\x87\xf1\x00\x00\x7f\xf6\x00\x008\x1c\x00\x00\x0c \x00\x00\x03\xc0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")

# Load the raspberry pi logo into the framebuffer (the image is 32x32)
fb = framebuf.FrameBuffer(buffer, 32, 32, framebuf.MONO_HLSB)

WIDTH = 128
HEIGHT = 64

try:
    oled = SSD1306_I2C(WIDTH,HEIGHT,i2c0)    
except:
    pass

oled.fill(0)
    
    
# ===============================================================    
# channel=HX711.CHANNEL_A_128, capvalue=1010
# channel=HX711.CHANNEL_A_64, capvalue=505
hx711 = HX711(d_out=16, pd_sck=17, channel=HX711.CHANNEL_A_128, capvalue=1010)
hx711.PowerOn()
utime.sleep_ms(500)

# 自动设置清零值
hx711.AutoCalibrationValue()

# 手动设置清零值

#calval = hx711.ReadCount(HX711.CHANNEL_A_128)
#print('calval',calval)
#hx711.SetCalibrationValue(calval)

hx711.Info()
utime.sleep(1)


while True:
    weights = hx711.GetWeights()
    print('W:',weights)
    #val = hx711.GetChannelBValue()
    #print(val)
    oled.fill(0)
    oled.text("Weights:{:.2f}V".format(weights), 0, 0)
    oled.show() 
    utime.sleep_ms(100)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

驱动库hx711.py

import machine
from machine import Pin
import utime
from micropython import const


class HX711:
    CHANNEL_A_128 = const(1)
    CHANNEL_B_32 = const(2)
    CHANNEL_A_64 = const(3)
    
    def __init__(self, d_out, pd_sck, channel=CHANNEL_A_128, capvalue=429.50, interval=20, timeout=100):
        self.dk = Pin(d_out, Pin.OUT)
        self.do = Pin(pd_sck, Pin.IN)
        self.channel = channel
        self.timeout = timeout*1000
        self.interval = interval
        self.lastread = 0
        self.capvalue = capvalue
        self.calibration = 0
        self.cur_count = 0
        
    def AutoCalibrationValue(self):
        sums = 0
        trys = 0
        for i in range(50):
            count =self.ReadCount(self.channel)
            if (count != 0):
                sums += count
                trys += 1
                if trys == 10:
                    break
            else:
                utime.sleep_ms(10)
        print(sums, trys)
        count =int(sums/trys)
        self.SetCalibrationValue(count)
                        
    def SetCalibrationValue(self, cal):
        self.calibration = cal
        print('calibration:', self.calibration)
        
    def PowerOff(self):
        self.dk.value(1)
        utime.sleep_ms(100)
        
    def PowerOn(self):
        self.dk.value(1)
        utime.sleep_ms(10)
        self.dk.value(0)
        utime.sleep_ms(100)
        
    def ReadCount(self, channel=CHANNEL_A_128):
        if (utime.ticks_us()/1000) < self.lastread:
            return self.cur_count
        else:
            self.lastread = int((utime.ticks_us()/1000)+self.interval)
            
        self.dk.value(0)
        #utime.sleep_us(1)
        count = 0
        errtime = utime.ticks_us()
        while (self.do.value() == 1):
            start = utime.ticks_us() -  errtime
            if start > self.timeout:
                return 0
            pass
        
        for i in range(24):
            self.dk.value(1)
            utime.sleep_us(1)
            self.dk.value(0)
            count = count << 1;
            if self.do.value():
                count += 1
                
        # 第25个脉冲，下次转换通道A，增益128
        if channel >= CHANNEL_A_128:
            self.dk.value(1)
            utime.sleep_us(1)
            self.dk.value(0)
        if channel >= CHANNEL_B_32:
            self.dk.value(1)
            utime.sleep_us(1)
            self.dk.value(0)
        if channel >= CHANNEL_A_64:
            self.dk.value(1)
            utime.sleep_us(1)
            self.dk.value(0)
        #utime.sleep_us(1)
        count = count^0x800000
        self.cur_count = count
        return self.cur_count
    
    def GetWeights(self):
        count = self.ReadCount(self.channel)
        if count == 0:
            print('HX711 ReadCount Error.')
            return 0
        #print('count:', count, 'calibration:', self.calibration)
        return float((count-self.calibration)/self.capvalue)
    
    def GetChannelBValue(self):
        count = self.ReadCount(CHANNEL_B_32)
        if count == 0:
            print('HX711 ReadCount Error.')
            return 0
        return  count
    
    def Info(self):
        print('CapValue:', self.capvalue)
        print('Calibration:', self.calibration)
        if self.channel == CHANNEL_A_128:
            print('Channel:A 128')
        elif self.channel == CHANNEL_A_64:
            print('Channel:A 64')        
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116