m5/Core2forAWS-MicroPython
0
0
Fork 0
mirror of https://github.com/m5stack/Core2forAWS-MicroPython.git synced 2026-05-20 10:30:31 -07:00
Code Issues Packages Projects Releases Wiki Activity

all: Reformat C and Python source code with tools/codeformat.py.

Browse Source 
This is run with uncrustify 0.70.1, and black 19.10b0.
This commit is contained in:
Damien George
2020-02-27 15:36:53 +11:00
parent 3f39d18c2b
commit 69661f3343
539 changed files with 10490 additions and 8248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drivers/dht/dht.py
+5 -2
View File
@@ -6,6 +6,7 @@ try:
except:
from pyb import dht_readinto
class DHTBase:
def __init__(self, pin):
self.pin = pin
@@ -14,9 +15,10 @@ class DHTBase:
def measure(self):
buf = self.buf
dht_readinto(self.pin, buf)
if (buf[0] + buf[1] + buf[2] + buf[3]) & 0xff != buf[4]:
if (buf[0] + buf[1] + buf[2] + buf[3]) & 0xFF != buf[4]:
raise Exception("checksum error")
class DHT11(DHTBase):
def humidity(self):
return self.buf[0]
@@ -24,12 +26,13 @@ class DHT11(DHTBase):
def temperature(self):
return self.buf[2]
class DHT22(DHTBase):
def humidity(self):
return (self.buf[0] << 8 | self.buf[1]) * 0.1
def temperature(self):
t = ((self.buf[2] & 0x7f) << 8 | self.buf[3]) * 0.1
t = ((self.buf[2] & 0x7F) << 8 | self.buf[3]) * 0.1
if self.buf[2] & 0x80:
t = -t
return t
drivers/display/lcd160cr.py
+72 -64
View File
@@ -29,16 +29,17 @@ _uart_baud_table = {
460800: 8,
}
class LCD160CR:
def __init__(self, connect=None, *, pwr=None, i2c=None, spi=None, i2c_addr=98):
if connect in ('X', 'Y', 'XY', 'YX'):
if connect in ("X", "Y", "XY", "YX"):
i = connect[-1]
j = connect[0]
y = j + '4'
elif connect == 'C':
y = j + "4"
elif connect == "C":
i = 2
j = 2
y = 'A7'
y = "A7"
else:
if pwr is None or i2c is None or spi is None:
raise ValueError('must specify valid "connect" or all of "pwr", "i2c" and "spi"')
@@ -74,8 +75,8 @@ class LCD160CR:
# set default orientation and window
self.set_orient(PORTRAIT)
self._fcmd2b('<BBBBBB', 0x76, 0, 0, self.w, self.h) # viewport 'v'
self._fcmd2b('<BBBBBB', 0x79, 0, 0, self.w, self.h) # window 'y'
self._fcmd2b("<BBBBBB", 0x76, 0, 0, self.w, self.h) # viewport 'v'
self._fcmd2b("<BBBBBB", 0x79, 0, 0, self.w, self.h) # window 'y'
def _send(self, cmd):
i = self.i2c.writeto(self.i2c_addr, cmd)
@@ -135,7 +136,7 @@ class LCD160CR:
@staticmethod
def rgb(r, g, b):
return ((b & 0xf8) << 8) | ((g & 0xfc) << 3) | (r >> 3)
return ((b & 0xF8) << 8) | ((g & 0xFC) << 3) | (r >> 3)
@staticmethod
def clip_line(c, w, h):
@@ -207,43 +208,43 @@ class LCD160CR:
sleep_ms(15)
def set_orient(self, orient):
self._fcmd2('<BBB', 0x14, (orient & 3) + 4)
self._fcmd2("<BBB", 0x14, (orient & 3) + 4)
# update width and height variables
self.iflush()
self._send(b'\x02g0')
self._send(b"\x02g0")
self._waitfor(4, self.buf[5])
self.w = self.buf[5][1]
self.h = self.buf[5][2]
def set_brightness(self, value):
self._fcmd2('<BBB', 0x16, value)
self._fcmd2("<BBB", 0x16, value)
def set_i2c_addr(self, addr):
# 0x0e set i2c addr
if addr & 3:
raise ValueError('must specify mod 4 aligned address')
self._fcmd2('<BBW', 0x0e, 0x433249 | (addr << 24))
raise ValueError("must specify mod 4 aligned address")
self._fcmd2("<BBW", 0x0E, 0x433249 | (addr << 24))
def set_uart_baudrate(self, baudrate):
try:
baudrate = _uart_baud_table[baudrate]
except KeyError:
raise ValueError('invalid baudrate')
self._fcmd2('<BBB', 0x18, baudrate)
raise ValueError("invalid baudrate")
self._fcmd2("<BBB", 0x18, baudrate)
def set_startup_deco(self, value):
self._fcmd2('<BBB', 0x19, value)
self._fcmd2("<BBB", 0x19, value)
def save_to_flash(self):
self._send(b'\x02fn')
self._send(b"\x02fn")
#### PIXEL ACCESS ####
def set_pixel(self, x, y, c):
self._fcmd2b('<BBBBH', 0x41, x, y, c)
self._fcmd2b("<BBBBH", 0x41, x, y, c)
def get_pixel(self, x, y):
self._fcmd2('<BBBB', 0x61, x, y)
self._fcmd2("<BBBB", 0x61, x, y)
t = 1000
while t:
self.i2c.readfrom_into(self.i2c_addr, self.buf1)
@@ -256,7 +257,7 @@ class LCD160CR:
def get_line(self, x, y, buf):
l = len(buf) // 2
self._fcmd2b('<BBBBB', 0x10, l, x, y)
self._fcmd2b("<BBBBB", 0x10, l, x, y)
l *= 2
t = 1000
while t:
@@ -280,42 +281,47 @@ class LCD160CR:
# split line if more than 254 bytes needed
buflen = (w + 1) // 2
line = bytearray(2 * buflen + 1)
line2 = memoryview(line)[:2 * (w - buflen) + 1]
line2 = memoryview(line)[: 2 * (w - buflen) + 1]
for i in range(min(len(buf) // (2 * w), h)):
ix = i * w * 2
self.get_line(x, y + i, line)
buf[ix:ix + len(line) - 1] = memoryview(line)[1:]
buf[ix : ix + len(line) - 1] = memoryview(line)[1:]
ix += len(line) - 1
if line2:
self.get_line(x + buflen, y + i, line2)
buf[ix:ix + len(line2) - 1] = memoryview(line2)[1:]
buf[ix : ix + len(line2) - 1] = memoryview(line2)[1:]
ix += len(line2) - 1
def screen_load(self, buf):
l = self.w * self.h * 2+2
self._fcmd2b('<BBHBBB', 0x70, l, 16, self.w, self.h)
l = self.w * self.h * 2 + 2
self._fcmd2b("<BBHBBB", 0x70, l, 16, self.w, self.h)
n = 0
ar = memoryview(buf)
while n < len(buf):
if len(buf) - n >= 0x200:
self._send(ar[n:n + 0x200])
self._send(ar[n : n + 0x200])
n += 0x200
else:
self._send(ar[n:])
while n < self.w * self.h * 2:
self._send(b'\x00')
self._send(b"\x00")
n += 1
#### TEXT COMMANDS ####
def set_pos(self, x, y):
self._fcmd2('<BBBB', 0x58, x, y)
self._fcmd2("<BBBB", 0x58, x, y)
def set_text_color(self, fg, bg):
self._fcmd2('<BBHH', 0x63, fg, bg)
self._fcmd2("<BBHH", 0x63, fg, bg)
def set_font(self, font, scale=0, bold=0, trans=0, scroll=0):
self._fcmd2('<BBBB', 0x46, (scroll << 7) | (trans << 6) | ((font & 3) << 4) | (bold & 0xf), scale & 0xff)
self._fcmd2(
"<BBBB",
0x46,
(scroll << 7) | (trans << 6) | ((font & 3) << 4) | (bold & 0xF),
scale & 0xFF,
)
def write(self, s):
# TODO: eventually check for room in LCD input queue
@@ -324,14 +330,14 @@ class LCD160CR:
#### PRIMITIVE DRAWING COMMANDS ####
def set_pen(self, line, fill):
self._fcmd2('<BBHH', 0x50, line, fill)
self._fcmd2("<BBHH", 0x50, line, fill)
def erase(self):
self._send(b'\x02\x45')
self._send(b"\x02\x45")
def dot(self, x, y):
if 0 <= x < self.w and 0 <= y < self.h:
self._fcmd2('<BBBB', 0x4b, x, y)
self._fcmd2("<BBBB", 0x4B, x, y)
def rect(self, x, y, w, h, cmd=0x72):
if x + w <= 0 or y + h <= 0 or x >= self.w or y >= self.h:
@@ -348,19 +354,19 @@ class LCD160CR:
y = 0
if cmd == 0x51 or cmd == 0x72:
# draw interior
self._fcmd2b('<BBBBBB', 0x51, x, y, min(w, 255), min(h, 255))
self._fcmd2b("<BBBBBB", 0x51, x, y, min(w, 255), min(h, 255))
if cmd == 0x57 or cmd == 0x72:
# draw outline
if left:
self._fcmd2b('<BBBBBB', 0x57, x, y, 1, min(h, 255))
self._fcmd2b("<BBBBBB", 0x57, x, y, 1, min(h, 255))
if top:
self._fcmd2b('<BBBBBB', 0x57, x, y, min(w, 255), 1)
self._fcmd2b("<BBBBBB", 0x57, x, y, min(w, 255), 1)
if x + w < self.w:
self._fcmd2b('<BBBBBB', 0x57, x + w, y, 1, min(h, 255))
self._fcmd2b("<BBBBBB", 0x57, x + w, y, 1, min(h, 255))
if y + h < self.h:
self._fcmd2b('<BBBBBB', 0x57, x, y + h, min(w, 255), 1)
self._fcmd2b("<BBBBBB", 0x57, x, y + h, min(w, 255), 1)
else:
self._fcmd2b('<BBBBBB', cmd, x, y, min(w, 255), min(h, 255))
self._fcmd2b("<BBBBBB", cmd, x, y, min(w, 255), min(h, 255))
def rect_outline(self, x, y, w, h):
self.rect(x, y, w, h, 0x57)
@@ -375,48 +381,48 @@ class LCD160CR:
ar4[2] = x2
ar4[3] = y2
if self.clip_line(ar4, self.w, self.h):
self._fcmd2b('<BBBBBB', 0x4c, ar4[0], ar4[1], ar4[2], ar4[3])
self._fcmd2b("<BBBBBB", 0x4C, ar4[0], ar4[1], ar4[2], ar4[3])
def dot_no_clip(self, x, y):
self._fcmd2('<BBBB', 0x4b, x, y)
self._fcmd2("<BBBB", 0x4B, x, y)
def rect_no_clip(self, x, y, w, h):
self._fcmd2b('<BBBBBB', 0x72, x, y, w, h)
self._fcmd2b("<BBBBBB", 0x72, x, y, w, h)
def rect_outline_no_clip(self, x, y, w, h):
self._fcmd2b('<BBBBBB', 0x57, x, y, w, h)
self._fcmd2b("<BBBBBB", 0x57, x, y, w, h)
def rect_interior_no_clip(self, x, y, w, h):
self._fcmd2b('<BBBBBB', 0x51, x, y, w, h)
self._fcmd2b("<BBBBBB", 0x51, x, y, w, h)
def line_no_clip(self, x1, y1, x2, y2):
self._fcmd2b('<BBBBBB', 0x4c, x1, y1, x2, y2)
self._fcmd2b("<BBBBBB", 0x4C, x1, y1, x2, y2)
def poly_dot(self, data):
if len(data) & 1:
raise ValueError('must specify even number of bytes')
self._fcmd2('<BBB', 0x71, len(data) // 2)
raise ValueError("must specify even number of bytes")
self._fcmd2("<BBB", 0x71, len(data) // 2)
self._send(data)
def poly_line(self, data):
if len(data) & 1:
raise ValueError('must specify even number of bytes')
self._fcmd2('<BBB', 0x78, len(data) // 2)
raise ValueError("must specify even number of bytes")
self._fcmd2("<BBB", 0x78, len(data) // 2)
self._send(data)
#### TOUCH COMMANDS ####
def touch_config(self, calib=False, save=False, irq=None):
self._fcmd2('<BBBB', 0x7a, (irq is not None) << 2 | save << 1 | calib, bool(irq) << 7)
self._fcmd2("<BBBB", 0x7A, (irq is not None) << 2 | save << 1 | calib, bool(irq) << 7)
def is_touched(self):
self._send(b'\x02T')
self._send(b"\x02T")
b = self.buf[4]
self._waitfor(3, b)
return b[1] >> 7 != 0
def get_touch(self):
self._send(b'\x02T') # implicit LCD output flush
self._send(b"\x02T") # implicit LCD output flush
b = self.buf[4]
self._waitfor(3, b)
return b[1] >> 7, b[2], b[3]
@@ -424,11 +430,13 @@ class LCD160CR:
#### ADVANCED COMMANDS ####
def set_spi_win(self, x, y, w, h):
pack_into('<BBBHHHHHHHH', self.buf19, 0, 2, 0x55, 10, x, y, x + w - 1, y + h - 1, 0, 0, 0, 0xffff)
pack_into(
"<BBBHHHHHHHH", self.buf19, 0, 2, 0x55, 10, x, y, x + w - 1, y + h - 1, 0, 0, 0, 0xFFFF
)
self._send(self.buf19)
def fast_spi(self, flush=True):
self._send(b'\x02\x12')
self._send(b"\x02\x12")
if flush:
self.oflush()
return self.spi
@@ -437,27 +445,27 @@ class LCD160CR:
self.fast_spi().write(buf)
def set_scroll(self, on):
self._fcmd2('<BBB', 0x15, on)
self._fcmd2("<BBB", 0x15, on)
def set_scroll_win(self, win, x=-1, y=0, w=0, h=0, vec=0, pat=0, fill=0x07e0, color=0):
pack_into('<BBBHHHHHHHH', self.buf19, 0, 2, 0x55, win, x, y, w, h, vec, pat, fill, color)
def set_scroll_win(self, win, x=-1, y=0, w=0, h=0, vec=0, pat=0, fill=0x07E0, color=0):
pack_into("<BBBHHHHHHHH", self.buf19, 0, 2, 0x55, win, x, y, w, h, vec, pat, fill, color)
self._send(self.buf19)
def set_scroll_win_param(self, win, param, value):
self._fcmd2b('<BBBBH', 0x75, win, param, value)
self._fcmd2b("<BBBBH", 0x75, win, param, value)
def set_scroll_buf(self, s):
l = len(s)
if l > 32:
raise ValueError('length must be 32 or less')
self._fcmd2('<BBB', 0x11, l)
raise ValueError("length must be 32 or less")
self._fcmd2("<BBB", 0x11, l)
self._send(s)
def jpeg_start(self, l):
if l > 0xffff:
raise ValueError('length must be 65535 or less')
if l > 0xFFFF:
raise ValueError("length must be 65535 or less")
self.oflush()
self._fcmd2('<BBH', 0x6a, l)
self._fcmd2("<BBH", 0x6A, l)
def jpeg_data(self, buf):
self._send(buf)
@@ -467,8 +475,8 @@ class LCD160CR:
self.jpeg_data(buf)
def feed_wdt(self):
self._send(b'\x02\x17')
self._send(b"\x02\x17")
def reset(self):
self._send(b'\x02Y\xef\xbe\xad\xde')
self._send(b"\x02Y\xef\xbe\xad\xde")
sleep_ms(15)
drivers/display/lcd160cr_test.py
+38 -22
View File
@@ -3,11 +3,13 @@
import time, math, framebuf, lcd160cr
def get_lcd(lcd):
if type(lcd) is str:
lcd = lcd160cr.LCD160CR(lcd)
return lcd
def show_adc(lcd, adc):
data = [adc.read_core_temp(), adc.read_core_vbat(), 3.3]
try:
@@ -21,24 +23,26 @@ def show_adc(lcd, adc):
lcd.set_pos(0, 100 + i * 16)
else:
lcd.set_font(2, trans=1)
lcd.set_pos(0, lcd.h-60 + i * 16)
lcd.write('%4s: ' % ('TEMP', 'VBAT', 'VREF')[i])
lcd.set_pos(0, lcd.h - 60 + i * 16)
lcd.write("%4s: " % ("TEMP", "VBAT", "VREF")[i])
if i > 0:
s = '%6.3fV' % data[i]
s = "%6.3fV" % data[i]
else:
s = '%5.1f°C' % data[i]
s = "%5.1f°C" % data[i]
if lcd.h == 160:
lcd.set_font(1, bold=0, scale=1)
else:
lcd.set_font(1, bold=0, scale=1, trans=1)
lcd.set_pos(45, lcd.h-60 + i * 16)
lcd.set_pos(45, lcd.h - 60 + i * 16)
lcd.write(s)
def test_features(lcd, orient=lcd160cr.PORTRAIT):
# if we run on pyboard then use ADC and RTC features
try:
import pyb
adc = pyb.ADCAll(12, 0xf0000)
adc = pyb.ADCAll(12, 0xF0000)
rtc = pyb.RTC()
except:
adc = None
@@ -53,7 +57,7 @@ def test_features(lcd, orient=lcd160cr.PORTRAIT):
# create M-logo
mlogo = framebuf.FrameBuffer(bytearray(17 * 17 * 2), 17, 17, framebuf.RGB565)
mlogo.fill(0)
mlogo.fill_rect(1, 1, 15, 15, 0xffffff)
mlogo.fill_rect(1, 1, 15, 15, 0xFFFFFF)
mlogo.vline(4, 4, 12, 0)
mlogo.vline(8, 1, 12, 0)
mlogo.vline(12, 4, 12, 0)
@@ -80,15 +84,18 @@ def test_features(lcd, orient=lcd160cr.PORTRAIT):
if tx2 >= 0 and ty2 >= 0 and tx2 < w and ty2 < h:
tx, ty = tx2, ty2
else:
tx = (tx + 1) % w
ty = (ty + 1) % h
tx = (tx + 1) % w
ty = (ty + 1) % h
# create and show the inline framebuf
fbuf.fill(lcd.rgb(128 + int(64 * math.cos(0.1 * i)), 128, 192))
fbuf.line(w // 2, h // 2,
fbuf.line(
w // 2,
h // 2,
w // 2 + int(40 * math.cos(0.2 * i)),
h // 2 + int(40 * math.sin(0.2 * i)),
lcd.rgb(128, 255, 64))
lcd.rgb(128, 255, 64),
)
fbuf.hline(0, ty, w, lcd.rgb(64, 64, 64))
fbuf.vline(tx, 0, h, lcd.rgb(64, 64, 64))
fbuf.rect(tx - 3, ty - 3, 7, 7, lcd.rgb(64, 64, 64))
@@ -97,9 +104,12 @@ def test_features(lcd, orient=lcd160cr.PORTRAIT):
y = h // 2 - 8 + int(32 * math.sin(0.05 * i + phase))
fbuf.blit(mlogo, x, y)
for j in range(-3, 3):
fbuf.text('MicroPython',
5, h // 2 + 9 * j + int(20 * math.sin(0.1 * (i + j))),
lcd.rgb(128 + 10 * j, 0, 128 - 10 * j))
fbuf.text(
"MicroPython",
5,
h // 2 + 9 * j + int(20 * math.sin(0.1 * (i + j))),
lcd.rgb(128 + 10 * j, 0, 128 - 10 * j),
)
lcd.show_framebuf(fbuf)
# show results from the ADC
@@ -111,7 +121,10 @@ def test_features(lcd, orient=lcd160cr.PORTRAIT):
lcd.set_pos(2, 0)
lcd.set_font(1)
t = rtc.datetime()
lcd.write('%4d-%02d-%02d %2d:%02d:%02d.%01d' % (t[0], t[1], t[2], t[4], t[5], t[6], t[7] // 100000))
lcd.write(
"%4d-%02d-%02d %2d:%02d:%02d.%01d"
% (t[0], t[1], t[2], t[4], t[5], t[6], t[7] // 100000)
)
# compute the frame rate
t1 = time.ticks_us()
@@ -120,13 +133,14 @@ def test_features(lcd, orient=lcd160cr.PORTRAIT):
# show the frame rate
lcd.set_pos(2, 9)
lcd.write('%.2f fps' % (1000000 / dt))
lcd.write("%.2f fps" % (1000000 / dt))
def test_mandel(lcd, orient=lcd160cr.PORTRAIT):
# set orientation and clear screen
lcd = get_lcd(lcd)
lcd.set_orient(orient)
lcd.set_pen(0, 0xffff)
lcd.set_pen(0, 0xFFFF)
lcd.erase()
# function to compute Mandelbrot pixels
@@ -148,24 +162,26 @@ def test_mandel(lcd, orient=lcd160cr.PORTRAIT):
spi = lcd.fast_spi()
# draw the Mandelbrot set line-by-line
hh = ((h - 1) / 3.2)
ww = ((w - 1) / 2.4)
hh = (h - 1) / 3.2
ww = (w - 1) / 2.4
for v in range(h):
for u in range(w):
c = in_set((v / hh - 2.3) + (u / ww - 1.2) * 1j)
if c < 16:
rgb = c << 12 | c << 6
else:
rgb = 0xf800 | c << 6
rgb = 0xF800 | c << 6
line[2 * u] = rgb
line[2 * u + 1] = rgb >> 8
spi.write(line)
def test_all(lcd, orient=lcd160cr.PORTRAIT):
lcd = get_lcd(lcd)
test_features(lcd, orient)
test_mandel(lcd, orient)
print('To run all tests: test_all(<lcd>)')
print('Individual tests are: test_features, test_mandel')
print("To run all tests: test_all(<lcd>)")
print("Individual tests are: test_features, test_mandel")
print('<lcd> argument should be a connection, eg "X", or an LCD160CR object')
drivers/display/ssd1306.py
+45 -34
View File
@@ -5,23 +5,23 @@ 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_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)
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
@@ -37,27 +37,37 @@ class SSD1306(framebuf.FrameBuffer):
def init_display(self):
for cmd in (
SET_DISP | 0x00, # off
SET_DISP | 0x00, # off
# address setting
SET_MEM_ADDR, 0x00, # horizontal
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.height == 32 else 0x12,
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.height == 32 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
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
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
SET_CHARGE_PUMP,
0x10 if self.external_vcc else 0x14,
SET_DISP | 0x01,
): # on
self.write_cmd(cmd)
self.fill(0)
self.show()
@@ -92,15 +102,15 @@ class SSD1306(framebuf.FrameBuffer):
class SSD1306_I2C(SSD1306):
def __init__(self, width, height, i2c, addr=0x3c, external_vcc=False):
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
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[0] = 0x80 # Co=1, D/C#=0
self.temp[1] = cmd
self.i2c.writeto(self.addr, self.temp)
@@ -120,6 +130,7 @@ class SSD1306_SPI(SSD1306):
self.res = res
self.cs = cs
import time
self.res(1)
time.sleep_ms(1)
self.res(0)
drivers/nrf24l01/nrf24l01.py
+38 -37
View File
@@ -5,49 +5,50 @@ from micropython import const
import utime
# nRF24L01+ registers
CONFIG = const(0x00)
EN_RXADDR = const(0x02)
SETUP_AW = const(0x03)
SETUP_RETR = const(0x04)
RF_CH = const(0x05)
RF_SETUP = const(0x06)
STATUS = const(0x07)
RX_ADDR_P0 = const(0x0a)
TX_ADDR = const(0x10)
RX_PW_P0 = const(0x11)
CONFIG = const(0x00)
EN_RXADDR = const(0x02)
SETUP_AW = const(0x03)
SETUP_RETR = const(0x04)
RF_CH = const(0x05)
RF_SETUP = const(0x06)
STATUS = const(0x07)
RX_ADDR_P0 = const(0x0A)
TX_ADDR = const(0x10)
RX_PW_P0 = const(0x11)
FIFO_STATUS = const(0x17)
DYNPD = const(0x1c)
DYNPD = const(0x1C)
# CONFIG register
EN_CRC = const(0x08) # enable CRC
CRCO = const(0x04) # CRC encoding scheme; 0=1 byte, 1=2 bytes
PWR_UP = const(0x02) # 1=power up, 0=power down
PRIM_RX = const(0x01) # RX/TX control; 0=PTX, 1=PRX
EN_CRC = const(0x08) # enable CRC
CRCO = const(0x04) # CRC encoding scheme; 0=1 byte, 1=2 bytes
PWR_UP = const(0x02) # 1=power up, 0=power down
PRIM_RX = const(0x01) # RX/TX control; 0=PTX, 1=PRX
# RF_SETUP register
POWER_0 = const(0x00) # -18 dBm
POWER_1 = const(0x02) # -12 dBm
POWER_2 = const(0x04) # -6 dBm
POWER_3 = const(0x06) # 0 dBm
SPEED_1M = const(0x00)
SPEED_2M = const(0x08)
SPEED_250K = const(0x20)
POWER_0 = const(0x00) # -18 dBm
POWER_1 = const(0x02) # -12 dBm
POWER_2 = const(0x04) # -6 dBm
POWER_3 = const(0x06) # 0 dBm
SPEED_1M = const(0x00)
SPEED_2M = const(0x08)
SPEED_250K = const(0x20)
# STATUS register
RX_DR = const(0x40) # RX data ready; write 1 to clear
TX_DS = const(0x20) # TX data sent; write 1 to clear
MAX_RT = const(0x10) # max retransmits reached; write 1 to clear
RX_DR = const(0x40) # RX data ready; write 1 to clear
TX_DS = const(0x20) # TX data sent; write 1 to clear
MAX_RT = const(0x10) # max retransmits reached; write 1 to clear
# FIFO_STATUS register
RX_EMPTY = const(0x01) # 1 if RX FIFO is empty
RX_EMPTY = const(0x01) # 1 if RX FIFO is empty
# constants for instructions
R_RX_PL_WID = const(0x60) # read RX payload width
R_RX_PAYLOAD = const(0x61) # read RX payload
W_TX_PAYLOAD = const(0xa0) # write TX payload
FLUSH_TX = const(0xe1) # flush TX FIFO
FLUSH_RX = const(0xe2) # flush RX FIFO
NOP = const(0xff) # use to read STATUS register
R_RX_PL_WID = const(0x60) # read RX payload width
R_RX_PAYLOAD = const(0x61) # read RX payload
W_TX_PAYLOAD = const(0xA0) # write TX payload
FLUSH_TX = const(0xE1) # flush TX FIFO
FLUSH_RX = const(0xE2) # flush RX FIFO
NOP = const(0xFF) # use to read STATUS register
class NRF24L01:
def __init__(self, spi, cs, ce, channel=46, payload_size=16):
@@ -84,7 +85,7 @@ class NRF24L01:
self.reg_write(SETUP_RETR, (6 << 4) | 8)
# set rf power and speed
self.set_power_speed(POWER_3, SPEED_250K) # Best for point to point links
self.set_power_speed(POWER_3, SPEED_250K) # Best for point to point links
# init CRC
self.set_crc(2)
@@ -218,7 +219,7 @@ class NRF24L01:
start = utime.ticks_ms()
result = None
while result is None and utime.ticks_diff(utime.ticks_ms(), start) < timeout:
result = self.send_done() # 1 == success, 2 == fail
result = self.send_done() # 1 == success, 2 == fail
if result == 2:
raise OSError("send failed")
@@ -232,18 +233,18 @@ class NRF24L01:
self.spi.readinto(self.buf, W_TX_PAYLOAD)
self.spi.write(buf)
if len(buf) < self.payload_size:
self.spi.write(b'\x00' * (self.payload_size - len(buf))) # pad out data
self.spi.write(b"\x00" * (self.payload_size - len(buf))) # pad out data
self.cs(1)
# enable the chip so it can send the data
self.ce(1)
utime.sleep_us(15) # needs to be >10us
utime.sleep_us(15) # needs to be >10us
self.ce(0)
# returns None if send still in progress, 1 for success, 2 for fail
def send_done(self):
if not (self.reg_read(STATUS) & (TX_DS | MAX_RT)):
return None # tx not finished
return None # tx not finished
# either finished or failed: get and clear status flags, power down
status = self.reg_write(STATUS, RX_DR | TX_DS | MAX_RT)
drivers/nrf24l01/nrf24l01test.py
+49 -39
View File
@@ -14,27 +14,28 @@ _RX_POLL_DELAY = const(15)
# master may be a slow device. Value tested with Pyboard, ESP32 and ESP8266.
_SLAVE_SEND_DELAY = const(10)
if sys.platform == 'pyboard':
cfg = {'spi': 2, 'miso': 'Y7', 'mosi': 'Y8', 'sck': 'Y6', 'csn': 'Y5', 'ce': 'Y4'}
elif sys.platform == 'esp8266': # Hardware SPI
cfg = {'spi': 1, 'miso': 12, 'mosi': 13, 'sck': 14, 'csn': 4, 'ce': 5}
elif sys.platform == 'esp32': # Software SPI
cfg = {'spi': -1, 'miso': 32, 'mosi': 33, 'sck': 25, 'csn': 26, 'ce': 27}
if sys.platform == "pyboard":
cfg = {"spi": 2, "miso": "Y7", "mosi": "Y8", "sck": "Y6", "csn": "Y5", "ce": "Y4"}
elif sys.platform == "esp8266": # Hardware SPI
cfg = {"spi": 1, "miso": 12, "mosi": 13, "sck": 14, "csn": 4, "ce": 5}
elif sys.platform == "esp32": # Software SPI
cfg = {"spi": -1, "miso": 32, "mosi": 33, "sck": 25, "csn": 26, "ce": 27}
else:
raise ValueError('Unsupported platform {}'.format(sys.platform))
raise ValueError("Unsupported platform {}".format(sys.platform))
# Addresses are in little-endian format. They correspond to big-endian
# 0xf0f0f0f0e1, 0xf0f0f0f0d2
pipes = (b'\xe1\xf0\xf0\xf0\xf0', b'\xd2\xf0\xf0\xf0\xf0')
pipes = (b"\xe1\xf0\xf0\xf0\xf0", b"\xd2\xf0\xf0\xf0\xf0")
def master():
csn = Pin(cfg['csn'], mode=Pin.OUT, value=1)
ce = Pin(cfg['ce'], mode=Pin.OUT, value=0)
if cfg['spi'] == -1:
spi = SPI(-1, sck=Pin(cfg['sck']), mosi=Pin(cfg['mosi']), miso=Pin(cfg['miso']))
csn = Pin(cfg["csn"], mode=Pin.OUT, value=1)
ce = Pin(cfg["ce"], mode=Pin.OUT, value=0)
if cfg["spi"] == -1:
spi = SPI(-1, sck=Pin(cfg["sck"]), mosi=Pin(cfg["mosi"]), miso=Pin(cfg["miso"]))
nrf = NRF24L01(spi, csn, ce, payload_size=8)
else:
nrf = NRF24L01(SPI(cfg['spi']), csn, ce, payload_size=8)
nrf = NRF24L01(SPI(cfg["spi"]), csn, ce, payload_size=8)
nrf.open_tx_pipe(pipes[0])
nrf.open_rx_pipe(1, pipes[1])
@@ -45,16 +46,16 @@ def master():
num_failures = 0
led_state = 0
print('NRF24L01 master mode, sending %d packets...' % num_needed)
print("NRF24L01 master mode, sending %d packets..." % num_needed)
while num_successes < num_needed and num_failures < num_needed:
# stop listening and send packet
nrf.stop_listening()
millis = utime.ticks_ms()
led_state = max(1, (led_state << 1) & 0x0f)
print('sending:', millis, led_state)
led_state = max(1, (led_state << 1) & 0x0F)
print("sending:", millis, led_state)
try:
nrf.send(struct.pack('ii', millis, led_state))
nrf.send(struct.pack("ii", millis, led_state))
except OSError:
pass
@@ -69,43 +70,50 @@ def master():
timeout = True
if timeout:
print('failed, response timed out')
print("failed, response timed out")
num_failures += 1
else:
# recv packet
got_millis, = struct.unpack('i', nrf.recv())
(got_millis,) = struct.unpack("i", nrf.recv())
# print response and round-trip delay
print('got response:', got_millis, '(delay', utime.ticks_diff(utime.ticks_ms(), got_millis), 'ms)')
print(
"got response:",
got_millis,
"(delay",
utime.ticks_diff(utime.ticks_ms(), got_millis),
"ms)",
)
num_successes += 1
# delay then loop
utime.sleep_ms(250)
print('master finished sending; successes=%d, failures=%d' % (num_successes, num_failures))
print("master finished sending; successes=%d, failures=%d" % (num_successes, num_failures))
def slave():
csn = Pin(cfg['csn'], mode=Pin.OUT, value=1)
ce = Pin(cfg['ce'], mode=Pin.OUT, value=0)
if cfg['spi'] == -1:
spi = SPI(-1, sck=Pin(cfg['sck']), mosi=Pin(cfg['mosi']), miso=Pin(cfg['miso']))
csn = Pin(cfg["csn"], mode=Pin.OUT, value=1)
ce = Pin(cfg["ce"], mode=Pin.OUT, value=0)
if cfg["spi"] == -1:
spi = SPI(-1, sck=Pin(cfg["sck"]), mosi=Pin(cfg["mosi"]), miso=Pin(cfg["miso"]))
nrf = NRF24L01(spi, csn, ce, payload_size=8)
else:
nrf = NRF24L01(SPI(cfg['spi']), csn, ce, payload_size=8)
nrf = NRF24L01(SPI(cfg["spi"]), csn, ce, payload_size=8)
nrf.open_tx_pipe(pipes[1])
nrf.open_rx_pipe(1, pipes[0])
nrf.start_listening()
print('NRF24L01 slave mode, waiting for packets... (ctrl-C to stop)')
print("NRF24L01 slave mode, waiting for packets... (ctrl-C to stop)")
while True:
if nrf.any():
while nrf.any():
buf = nrf.recv()
millis, led_state = struct.unpack('ii', buf)
print('received:', millis, led_state)
millis, led_state = struct.unpack("ii", buf)
print("received:", millis, led_state)
for led in leds:
if led_state & 1:
led.on()
@@ -118,23 +126,25 @@ def slave():
utime.sleep_ms(_SLAVE_SEND_DELAY)
nrf.stop_listening()
try:
nrf.send(struct.pack('i', millis))
nrf.send(struct.pack("i", millis))
except OSError:
pass
print('sent response')
print("sent response")
nrf.start_listening()
try:
import pyb
leds = [pyb.LED(i + 1) for i in range(4)]
except:
leds = []
print('NRF24L01 test module loaded')
print('NRF24L01 pinout for test:')
print(' CE on', cfg['ce'])
print(' CSN on', cfg['csn'])
print(' SCK on', cfg['sck'])
print(' MISO on', cfg['miso'])
print(' MOSI on', cfg['mosi'])
print('run nrf24l01test.slave() on slave, then nrf24l01test.master() on master')
print("NRF24L01 test module loaded")
print("NRF24L01 pinout for test:")
print(" CE on", cfg["ce"])
print(" CSN on", cfg["csn"])
print(" SCK on", cfg["sck"])
print(" MISO on", cfg["miso"])
print(" MOSI on", cfg["mosi"])
print("run nrf24l01test.slave() on slave, then nrf24l01test.master() on master")
drivers/onewire/ds18x20.py
+7 -6
View File
@@ -4,8 +4,9 @@
from micropython import const
_CONVERT = const(0x44)
_RD_SCRATCH = const(0xbe)
_WR_SCRATCH = const(0x4e)
_RD_SCRATCH = const(0xBE)
_WR_SCRATCH = const(0x4E)
class DS18X20:
def __init__(self, onewire):
@@ -26,7 +27,7 @@ class DS18X20:
self.ow.writebyte(_RD_SCRATCH)
self.ow.readinto(self.buf)
if self.ow.crc8(self.buf):
raise Exception('CRC error')
raise Exception("CRC error")
return self.buf
def write_scratch(self, rom, buf):
@@ -40,12 +41,12 @@ class DS18X20:
if rom[0] == 0x10:
if buf[1]:
t = buf[0] >> 1 | 0x80
t = -((~t + 1) & 0xff)
t = -((~t + 1) & 0xFF)
else:
t = buf[0] >> 1
return t - 0.25 + (buf[7] - buf[6]) / buf[7]
else:
t = buf[1] << 8 | buf[0]
if t & 0x8000: # sign bit set
t = -((t ^ 0xffff) + 1)
if t & 0x8000: # sign bit set
t = -((t ^ 0xFFFF) + 1)
return t / 16
drivers/onewire/onewire.py
+7 -5
View File
@@ -4,13 +4,15 @@
from micropython import const
import _onewire as _ow
class OneWireError(Exception):
pass
class OneWire:
SEARCH_ROM = const(0xf0)
SEARCH_ROM = const(0xF0)
MATCH_ROM = const(0x55)
SKIP_ROM = const(0xcc)
SKIP_ROM = const(0xCC)
def __init__(self, pin):
self.pin = pin
@@ -51,7 +53,7 @@ class OneWire:
devices = []
diff = 65
rom = False
for i in range(0xff):
for i in range(0xFF):
rom, diff = self._search_rom(rom, diff)
if rom:
devices += [rom]
@@ -73,10 +75,10 @@ class OneWire:
for bit in range(8):
b = self.readbit()
if self.readbit():
if b: # there are no devices or there is an error on the bus
if b: # there are no devices or there is an error on the bus
return None, 0
else:
if not b: # collision, two devices with different bit meaning
if not b: # collision, two devices with different bit meaning
if diff > i or ((l_rom[byte] & (1 << bit)) and diff != i):
b = 1
next_diff = i
drivers/sdcard/sdcard.py
+44 -44
View File
@@ -27,15 +27,15 @@ import time
_CMD_TIMEOUT = const(100)
_R1_IDLE_STATE = const(1 << 0)
#R1_ERASE_RESET = const(1 << 1)
# R1_ERASE_RESET = const(1 << 1)
_R1_ILLEGAL_COMMAND = const(1 << 2)
#R1_COM_CRC_ERROR = const(1 << 3)
#R1_ERASE_SEQUENCE_ERROR = const(1 << 4)
#R1_ADDRESS_ERROR = const(1 << 5)
#R1_PARAMETER_ERROR = const(1 << 6)
_TOKEN_CMD25 = const(0xfc)
_TOKEN_STOP_TRAN = const(0xfd)
_TOKEN_DATA = const(0xfe)
# R1_COM_CRC_ERROR = const(1 << 3)
# R1_ERASE_SEQUENCE_ERROR = const(1 << 4)
# R1_ADDRESS_ERROR = const(1 << 5)
# R1_PARAMETER_ERROR = const(1 << 6)
_TOKEN_CMD25 = const(0xFC)
_TOKEN_STOP_TRAN = const(0xFD)
_TOKEN_DATA = const(0xFE)
class SDCard:
@@ -47,7 +47,7 @@ class SDCard:
self.dummybuf = bytearray(512)
self.tokenbuf = bytearray(1)
for i in range(512):
self.dummybuf[i] = 0xff
self.dummybuf[i] = 0xFF
self.dummybuf_memoryview = memoryview(self.dummybuf)
# initialise the card
@@ -72,7 +72,7 @@ class SDCard:
# clock card at least 100 cycles with cs high
for i in range(16):
self.spi.write(b'\xff')
self.spi.write(b"\xff")
# CMD0: init card; should return _R1_IDLE_STATE (allow 5 attempts)
for _ in range(5):
@@ -82,7 +82,7 @@ class SDCard:
raise OSError("no SD card")
# CMD8: determine card version
r = self.cmd(8, 0x01aa, 0x87, 4)
r = self.cmd(8, 0x01AA, 0x87, 4)
if r == _R1_IDLE_STATE:
self.init_card_v2()
elif r == (_R1_IDLE_STATE | _R1_ILLEGAL_COMMAND):
@@ -96,15 +96,15 @@ class SDCard:
raise OSError("no response from SD card")
csd = bytearray(16)
self.readinto(csd)
if csd[0] & 0xc0 == 0x40: # CSD version 2.0
if csd[0] & 0xC0 == 0x40: # CSD version 2.0
self.sectors = ((csd[8] << 8 | csd[9]) + 1) * 1024
elif csd[0] & 0xc0 == 0x00: # CSD version 1.0 (old, <=2GB)
elif csd[0] & 0xC0 == 0x00: # CSD version 1.0 (old, <=2GB)
c_size = csd[6] & 0b11 | csd[7] << 2 | (csd[8] & 0b11000000) << 4
c_size_mult = ((csd[9] & 0b11) << 1) | csd[10] >> 7
self.sectors = (c_size + 1) * (2 ** (c_size_mult + 2))
else:
raise OSError("SD card CSD format not supported")
#print('sectors', self.sectors)
# print('sectors', self.sectors)
# CMD16: set block length to 512 bytes
if self.cmd(16, 512, 0) != 0:
@@ -118,7 +118,7 @@ class SDCard:
self.cmd(55, 0, 0)
if self.cmd(41, 0, 0) == 0:
self.cdv = 512
#print("[SDCard] v1 card")
# print("[SDCard] v1 card")
return
raise OSError("timeout waiting for v1 card")
@@ -130,7 +130,7 @@ class SDCard:
if self.cmd(41, 0x40000000, 0) == 0:
self.cmd(58, 0, 0, 4)
self.cdv = 1
#print("[SDCard] v2 card")
# print("[SDCard] v2 card")
return
raise OSError("timeout waiting for v2 card")
@@ -148,24 +148,24 @@ class SDCard:
self.spi.write(buf)
if skip1:
self.spi.readinto(self.tokenbuf, 0xff)
self.spi.readinto(self.tokenbuf, 0xFF)
# wait for the response (response[7] == 0)
for i in range(_CMD_TIMEOUT):
self.spi.readinto(self.tokenbuf, 0xff)
self.spi.readinto(self.tokenbuf, 0xFF)
response = self.tokenbuf[0]
if not (response & 0x80):
# this could be a big-endian integer that we are getting here
for j in range(final):
self.spi.write(b'\xff')
self.spi.write(b"\xff")
if release:
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
return response
# timeout
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
return -1
def readinto(self, buf):
@@ -173,7 +173,7 @@ class SDCard:
# read until start byte (0xff)
for i in range(_CMD_TIMEOUT):
self.spi.readinto(self.tokenbuf, 0xff)
self.spi.readinto(self.tokenbuf, 0xFF)
if self.tokenbuf[0] == _TOKEN_DATA:
break
else:
@@ -183,15 +183,15 @@ class SDCard:
# read data
mv = self.dummybuf_memoryview
if len(buf) != len(mv):
mv = mv[:len(buf)]
mv = mv[: len(buf)]
self.spi.write_readinto(mv, buf)
# read checksum
self.spi.write(b'\xff')
self.spi.write(b'\xff')
self.spi.write(b"\xff")
self.spi.write(b"\xff")
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
def write(self, token, buf):
self.cs(0)
@@ -199,42 +199,42 @@ class SDCard:
# send: start of block, data, checksum
self.spi.read(1, token)
self.spi.write(buf)
self.spi.write(b'\xff')
self.spi.write(b'\xff')
self.spi.write(b"\xff")
self.spi.write(b"\xff")
# check the response
if (self.spi.read(1, 0xff)[0] & 0x1f) != 0x05:
if (self.spi.read(1, 0xFF)[0] & 0x1F) != 0x05:
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
return
# wait for write to finish
while self.spi.read(1, 0xff)[0] == 0:
while self.spi.read(1, 0xFF)[0] == 0:
pass
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
def write_token(self, token):
self.cs(0)
self.spi.read(1, token)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
# wait for write to finish
while self.spi.read(1, 0xff)[0] == 0x00:
while self.spi.read(1, 0xFF)[0] == 0x00:
pass
self.cs(1)
self.spi.write(b'\xff')
self.spi.write(b"\xff")
def readblocks(self, block_num, buf):
nblocks = len(buf) // 512
assert nblocks and not len(buf) % 512, 'Buffer length is invalid'
assert nblocks and not len(buf) % 512, "Buffer length is invalid"
if nblocks == 1:
# CMD17: set read address for single block
if self.cmd(17, block_num * self.cdv, 0, release=False) != 0:
# release the card
self.cs(1)
raise OSError(5) # EIO
raise OSError(5) # EIO
# receive the data and release card
self.readinto(buf)
else:
@@ -242,7 +242,7 @@ class SDCard:
if self.cmd(18, block_num * self.cdv, 0, release=False) != 0:
# release the card
self.cs(1)
raise OSError(5) # EIO
raise OSError(5) # EIO
offset = 0
mv = memoryview(buf)
while nblocks:
@@ -250,23 +250,23 @@ class SDCard:
self.readinto(mv[offset : offset + 512])
offset += 512
nblocks -= 1
if self.cmd(12, 0, 0xff, skip1=True):
raise OSError(5) # EIO
if self.cmd(12, 0, 0xFF, skip1=True):
raise OSError(5) # EIO
def writeblocks(self, block_num, buf):
nblocks, err = divmod(len(buf), 512)
assert nblocks and not err, 'Buffer length is invalid'
assert nblocks and not err, "Buffer length is invalid"
if nblocks == 1:
# CMD24: set write address for single block
if self.cmd(24, block_num * self.cdv, 0) != 0:
raise OSError(5) # EIO
raise OSError(5) # EIO
# send the data
self.write(_TOKEN_DATA, buf)
else:
# CMD25: set write address for first block
if self.cmd(25, block_num * self.cdv, 0) != 0:
raise OSError(5) # EIO
raise OSError(5) # EIO
# send the data
offset = 0
mv = memoryview(buf)
@@ -277,5 +277,5 @@ class SDCard:
self.write_token(_TOKEN_STOP_TRAN)
def ioctl(self, op, arg):
if op == 4: # get number of blocks
if op == 4: # get number of blocks
return self.sectors
drivers/sdcard/sdtest.py
+31 -30
View File
@@ -2,59 +2,60 @@
# Peter hinch 30th Jan 2016
import os, sdcard, machine
def sdtest():
spi = machine.SPI(1)
spi.init() # Ensure right baudrate
sd = sdcard.SDCard(spi, machine.Pin.board.X21) # Compatible with PCB
sd = sdcard.SDCard(spi, machine.Pin.board.X21) # Compatible with PCB
vfs = os.VfsFat(sd)
os.mount(vfs, '/fc')
print('Filesystem check')
print(os.listdir('/fc'))
os.mount(vfs, "/fc")
print("Filesystem check")
print(os.listdir("/fc"))
line = 'abcdefghijklmnopqrstuvwxyz\n'
lines = line * 200 # 5400 chars
short = '1234567890\n'
line = "abcdefghijklmnopqrstuvwxyz\n"
lines = line * 200 # 5400 chars
short = "1234567890\n"
fn = '/fc/rats.txt'
fn = "/fc/rats.txt"
print()
print('Multiple block read/write')
with open(fn,'w') as f:
print("Multiple block read/write")
with open(fn, "w") as f:
n = f.write(lines)
print(n, 'bytes written')
print(n, "bytes written")
n = f.write(short)
print(n, 'bytes written')
print(n, "bytes written")
n = f.write(lines)
print(n, 'bytes written')
print(n, "bytes written")
with open(fn,'r') as f:
with open(fn, "r") as f:
result1 = f.read()
print(len(result1), 'bytes read')
print(len(result1), "bytes read")
fn = '/fc/rats1.txt'
fn = "/fc/rats1.txt"
print()
print('Single block read/write')
with open(fn,'w') as f:
n = f.write(short) # one block
print(n, 'bytes written')
print("Single block read/write")
with open(fn, "w") as f:
n = f.write(short) # one block
print(n, "bytes written")
with open(fn,'r') as f:
with open(fn, "r") as f:
result2 = f.read()
print(len(result2), 'bytes read')
print(len(result2), "bytes read")
os.umount('/fc')
os.umount("/fc")
print()
print('Verifying data read back')
print("Verifying data read back")
success = True
if result1 == ''.join((lines, short, lines)):
print('Large file Pass')
if result1 == "".join((lines, short, lines)):
print("Large file Pass")
else:
print('Large file Fail')
print("Large file Fail")
success = False
if result2 == short:
print('Small file Pass')
print("Small file Pass")
else:
print('Small file Fail')
print("Small file Fail")
success = False
print()
print('Tests', 'passed' if success else 'failed')
print("Tests", "passed" if success else "failed")
examples/SDdatalogger/boot.py
+10 -10
View File
@@ -8,18 +8,18 @@
import pyb
pyb.LED(3).on() # indicate we are waiting for switch press
pyb.delay(2000) # wait for user to maybe press the switch
switch_value = pyb.Switch()() # sample the switch at end of delay
pyb.LED(3).off() # indicate that we finished waiting for the switch
pyb.LED(3).on() # indicate we are waiting for switch press
pyb.delay(2000) # wait for user to maybe press the switch
switch_value = pyb.Switch()() # sample the switch at end of delay
pyb.LED(3).off() # indicate that we finished waiting for the switch
pyb.LED(4).on() # indicate that we are selecting the mode
pyb.LED(4).on() # indicate that we are selecting the mode
if switch_value:
pyb.usb_mode('VCP+MSC')
pyb.main('cardreader.py') # if switch was pressed, run this
pyb.usb_mode("VCP+MSC")
pyb.main("cardreader.py") # if switch was pressed, run this
else:
pyb.usb_mode('VCP+HID')
pyb.main('datalogger.py') # if switch wasn't pressed, run this
pyb.usb_mode("VCP+HID")
pyb.main("datalogger.py") # if switch wasn't pressed, run this
pyb.LED(4).off() # indicate that we finished selecting the mode
pyb.LED(4).off() # indicate that we finished selecting the mode
examples/SDdatalogger/datalogger.py
+9 -9
View File
@@ -17,17 +17,17 @@ while True:
# start if switch is pressed
if switch():
pyb.delay(200) # delay avoids detection of multiple presses
blue.on() # blue LED indicates file open
log = open('/sd/log.csv', 'w') # open file on SD (SD: '/sd/', flash: '/flash/)
pyb.delay(200) # delay avoids detection of multiple presses
blue.on() # blue LED indicates file open
log = open("/sd/log.csv", "w") # open file on SD (SD: '/sd/', flash: '/flash/)
# until switch is pressed again
while not switch():
t = pyb.millis() # get time
x, y, z = accel.filtered_xyz() # get acceleration data
log.write('{},{},{},{}\n'.format(t,x,y,z)) # write data to file
t = pyb.millis() # get time
x, y, z = accel.filtered_xyz() # get acceleration data
log.write("{},{},{},{}\n".format(t, x, y, z)) # write data to file
# end after switch is pressed again
log.close() # close file
blue.off() # blue LED indicates file closed
pyb.delay(200) # delay avoids detection of multiple presses
log.close() # close file
blue.off() # blue LED indicates file closed
pyb.delay(200) # delay avoids detection of multiple presses
examples/accel_i2c.py
+2 -2
View File
@@ -17,10 +17,10 @@ accel_pwr.value(1)
i2c = I2C(1, baudrate=100000)
addrs = i2c.scan()
print("Scanning devices:", [hex(x) for x in addrs])
if 0x4c not in addrs:
if 0x4C not in addrs:
print("Accelerometer is not detected")
ACCEL_ADDR = 0x4c
ACCEL_ADDR = 0x4C
ACCEL_AXIS_X_REG = 0
ACCEL_MODE_REG = 7
examples/accellog.py
+9 -9
View File
@@ -2,19 +2,19 @@
import pyb
accel = pyb.Accel() # create object of accelerometer
blue = pyb.LED(4) # create object of blue LED
accel = pyb.Accel() # create object of accelerometer
blue = pyb.LED(4) # create object of blue LED
# open file to write data - /sd/ is the SD-card, /flash/ the internal memory
log = open('/sd/log.csv', 'w')
log = open("/sd/log.csv", "w")
blue.on() # turn on blue LED
blue.on() # turn on blue LED
# do 100 times (if the board is connected via USB, you can't write longer because the PC tries to open the filesystem which messes up your file.)
for i in range(100):
t = pyb.millis() # get time since reset
x, y, z = accel.filtered_xyz() # get acceleration data
log.write('{},{},{},{}\n'.format(t,x,y,z)) # write data to file
t = pyb.millis() # get time since reset
x, y, z = accel.filtered_xyz() # get acceleration data
log.write("{},{},{},{}\n".format(t, x, y, z)) # write data to file
log.close() # close file
blue.off() # turn off LED
log.close() # close file
blue.off() # turn off LED
examples/asmled.py
+8 -6
View File
@@ -2,8 +2,8 @@
# this version is overly verbose and uses word stores
@micropython.asm_thumb
def flash_led(r0):
movw(r1, (stm.GPIOA + stm.GPIO_BSRRL) & 0xffff)
movt(r1, ((stm.GPIOA + stm.GPIO_BSRRL) >> 16) & 0x7fff)
movw(r1, (stm.GPIOA + stm.GPIO_BSRRL) & 0xFFFF)
movt(r1, ((stm.GPIOA + stm.GPIO_BSRRL) >> 16) & 0x7FFF)
movw(r2, 1 << 13)
movt(r2, 0)
movw(r3, 0)
@@ -17,8 +17,8 @@ def flash_led(r0):
str(r2, [r1, 0])
# delay for a bit
movw(r4, 5599900 & 0xffff)
movt(r4, (5599900 >> 16) & 0xffff)
movw(r4, 5599900 & 0xFFFF)
movt(r4, (5599900 >> 16) & 0xFFFF)
label(delay_on)
sub(r4, r4, 1)
cmp(r4, 0)
@@ -28,8 +28,8 @@ def flash_led(r0):
str(r3, [r1, 0])
# delay for a bit
movw(r4, 5599900 & 0xffff)
movt(r4, (5599900 >> 16) & 0xffff)
movw(r4, 5599900 & 0xFFFF)
movt(r4, (5599900 >> 16) & 0xFFFF)
label(delay_off)
sub(r4, r4, 1)
cmp(r4, 0)
@@ -41,6 +41,7 @@ def flash_led(r0):
cmp(r0, 0)
bgt(loop1)
# flash LED #2 using inline assembler
# this version uses half-word sortes, and the convenience assembler operation 'movwt'
@micropython.asm_thumb
@@ -81,5 +82,6 @@ def flash_led_v2(r0):
cmp(r0, 0)
bgt(loop1)
flash_led(5)
flash_led_v2(5)
examples/asmsum.py
+4 -2
View File
@@ -22,6 +22,7 @@ def asm_sum_words(r0, r1):
mov(r0, r2)
@micropython.asm_thumb
def asm_sum_bytes(r0, r1):
@@ -46,12 +47,13 @@ def asm_sum_bytes(r0, r1):
mov(r0, r2)
import array
b = array.array('l', (100, 200, 300, 400))
b = array.array("l", (100, 200, 300, 400))
n = asm_sum_words(len(b), b)
print(b, n)
b = array.array('b', (10, 20, 30, 40, 50, 60, 70, 80))
b = array.array("b", (10, 20, 30, 40, 50, 60, 70, 80))
n = asm_sum_bytes(len(b), b)
print(b, n)
examples/bluetooth/ble_advertising.py
+16 -9
View File
@@ -26,9 +26,12 @@ def advertising_payload(limited_disc=False, br_edr=False, name=None, services=No
def _append(adv_type, value):
nonlocal payload
payload += struct.pack('BB', len(value) + 1, adv_type) + value
payload += struct.pack("BB", len(value) + 1, adv_type) + value
_append(_ADV_TYPE_FLAGS, struct.pack('B', (0x01 if limited_disc else 0x02) + (0x00 if br_edr else 0x04)))
_append(
_ADV_TYPE_FLAGS,
struct.pack("B", (0x01 if limited_disc else 0x02) + (0x00 if br_edr else 0x04)),
)
if name:
_append(_ADV_TYPE_NAME, name)
@@ -44,7 +47,7 @@ def advertising_payload(limited_disc=False, br_edr=False, name=None, services=No
_append(_ADV_TYPE_UUID128_COMPLETE, b)
# See org.bluetooth.characteristic.gap.appearance.xml
_append(_ADV_TYPE_APPEARANCE, struct.pack('<h', appearance))
_append(_ADV_TYPE_APPEARANCE, struct.pack("<h", appearance))
return payload
@@ -54,32 +57,36 @@ def decode_field(payload, adv_type):
result = []
while i + 1 < len(payload):
if payload[i + 1] == adv_type:
result.append(payload[i + 2:i + payload[i] + 1])
result.append(payload[i + 2 : i + payload[i] + 1])
i += 1 + payload[i]
return result
def decode_name(payload):
n = decode_field(payload, _ADV_TYPE_NAME)
return str(n[0], 'utf-8') if n else ''
return str(n[0], "utf-8") if n else ""
def decode_services(payload):
services = []
for u in decode_field(payload, _ADV_TYPE_UUID16_COMPLETE):
services.append(bluetooth.UUID(struct.unpack('<h', u)[0]))
services.append(bluetooth.UUID(struct.unpack("<h", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID32_COMPLETE):
services.append(bluetooth.UUID(struct.unpack('<d', u)[0]))
services.append(bluetooth.UUID(struct.unpack("<d", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID128_COMPLETE):
services.append(bluetooth.UUID(u))
return services
def demo():
payload = advertising_payload(name='micropython', services=[bluetooth.UUID(0x181A), bluetooth.UUID('6E400001-B5A3-F393-E0A9-E50E24DCCA9E')])
payload = advertising_payload(
name="micropython",
services=[bluetooth.UUID(0x181A), bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")],
)
print(payload)
print(decode_name(payload))
print(decode_services(payload))
if __name__ == '__main__':
if __name__ == "__main__":
demo()
examples/bluetooth/ble_temperature.py
+18 -8
View File
@@ -10,26 +10,36 @@ import time
from ble_advertising import advertising_payload
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
# org.bluetooth.service.environmental_sensing
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# org.bluetooth.characteristic.temperature
_TEMP_CHAR = (bluetooth.UUID(0x2A6E), bluetooth.FLAG_READ|bluetooth.FLAG_NOTIFY,)
_ENV_SENSE_SERVICE = (_ENV_SENSE_UUID, (_TEMP_CHAR,),)
_TEMP_CHAR = (
bluetooth.UUID(0x2A6E),
bluetooth.FLAG_READ | bluetooth.FLAG_NOTIFY,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
class BLETemperature:
def __init__(self, ble, name='mpy-temp'):
def __init__(self, ble, name="mpy-temp"):
self._ble = ble
self._ble.active(True)
self._ble.irq(handler=self._irq)
((self._handle,),) = self._ble.gatts_register_services((_ENV_SENSE_SERVICE,))
self._connections = set()
self._payload = advertising_payload(name=name, services=[_ENV_SENSE_UUID], appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER)
self._payload = advertising_payload(
name=name, services=[_ENV_SENSE_UUID], appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER
)
self._advertise()
def _irq(self, event, data):
@@ -46,7 +56,7 @@ class BLETemperature:
def set_temperature(self, temp_deg_c, notify=False):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
# Write the local value, ready for a central to read.
self._ble.gatts_write(self._handle, struct.pack('<h', int(temp_deg_c * 100)))
self._ble.gatts_write(self._handle, struct.pack("<h", int(temp_deg_c * 100)))
if notify:
for conn_handle in self._connections:
# Notify connected centrals to issue a read.
@@ -72,5 +82,5 @@ def demo():
time.sleep_ms(1000)
if __name__ == '__main__':
if __name__ == "__main__":
demo()
examples/bluetooth/ble_temperature_central.py
+40 -28
View File
@@ -9,33 +9,41 @@ import micropython
from ble_advertising import decode_services, decode_name
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_IRQ_GATTS_READ_REQUEST = const(1 << 3)
_IRQ_SCAN_RESULT = const(1 << 4)
_IRQ_SCAN_COMPLETE = const(1 << 5)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_SERVICE_RESULT = const(1 << 8)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9)
_IRQ_GATTC_DESCRIPTOR_RESULT = const(1 << 10)
_IRQ_GATTC_READ_RESULT = const(1 << 11)
_IRQ_GATTC_WRITE_STATUS = const(1 << 12)
_IRQ_GATTC_NOTIFY = const(1 << 13)
_IRQ_GATTC_INDICATE = const(1 << 14)
_IRQ_ALL = const(0xffff)
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_IRQ_GATTS_READ_REQUEST = const(1 << 3)
_IRQ_SCAN_RESULT = const(1 << 4)
_IRQ_SCAN_COMPLETE = const(1 << 5)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_SERVICE_RESULT = const(1 << 8)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9)
_IRQ_GATTC_DESCRIPTOR_RESULT = const(1 << 10)
_IRQ_GATTC_READ_RESULT = const(1 << 11)
_IRQ_GATTC_WRITE_STATUS = const(1 << 12)
_IRQ_GATTC_NOTIFY = const(1 << 13)
_IRQ_GATTC_INDICATE = const(1 << 14)
_IRQ_ALL = const(0xFFFF)
# org.bluetooth.service.environmental_sensing
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# org.bluetooth.characteristic.temperature
_TEMP_UUID = bluetooth.UUID(0x2A6E)
_TEMP_CHAR = (_TEMP_UUID, bluetooth.FLAG_READ|bluetooth.FLAG_NOTIFY,)
_ENV_SENSE_SERVICE = (_ENV_SENSE_UUID, (_TEMP_CHAR,),)
_TEMP_CHAR = (
_TEMP_UUID,
bluetooth.FLAG_READ | bluetooth.FLAG_NOTIFY,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
class BLETemperatureCentral:
def __init__(self, ble):
self._ble = ble
@@ -72,8 +80,10 @@ class BLETemperatureCentral:
if connectable and _ENV_SENSE_UUID in decode_services(adv_data):
# Found a potential device, remember it and stop scanning.
self._addr_type = addr_type
self._addr = bytes(addr) # Note: addr buffer is owned by caller so need to copy it.
self._name = decode_name(adv_data) or '?'
self._addr = bytes(
addr
) # Note: addr buffer is owned by caller so need to copy it.
self._name = decode_name(adv_data) or "?"
self._ble.gap_scan(None)
elif event == _IRQ_SCAN_COMPLETE:
@@ -104,7 +114,9 @@ class BLETemperatureCentral:
# Connected device returned a service.
conn_handle, start_handle, end_handle, uuid = data
if conn_handle == self._conn_handle and uuid == _ENV_SENSE_UUID:
self._ble.gattc_discover_characteristics(self._conn_handle, start_handle, end_handle)
self._ble.gattc_discover_characteristics(
self._conn_handle, start_handle, end_handle
)
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
# Connected device returned a characteristic.
@@ -132,7 +144,6 @@ class BLETemperatureCentral:
if self._notify_callback:
self._notify_callback(self._value)
# Returns true if we've successfully connected and discovered characteristics.
def is_connected(self):
return self._conn_handle is not None and self._value_handle is not None
@@ -174,7 +185,7 @@ class BLETemperatureCentral:
def _update_value(self, data):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
self._value = struct.unpack('<h', data)[0] / 100
self._value = struct.unpack("<h", data)[0] / 100
return self._value
def value(self):
@@ -189,12 +200,12 @@ def demo():
def on_scan(addr_type, addr, name):
if addr_type is not None:
print('Found sensor:', addr_type, addr, name)
print("Found sensor:", addr_type, addr, name)
central.connect()
else:
nonlocal not_found
not_found = True
print('No sensor found.')
print("No sensor found.")
central.scan(callback=on_scan)
@@ -204,7 +215,7 @@ def demo():
if not_found:
return
print('Connected')
print("Connected")
# Explicitly issue reads, using "print" as the callback.
while central.is_connected():
@@ -216,7 +227,8 @@ def demo():
# print(central.value())
# time.sleep_ms(2000)
print('Disconnected')
print("Disconnected")
if __name__ == '__main__':
if __name__ == "__main__":
demo()
examples/bluetooth/ble_uart_peripheral.py
+25 -12
View File
@@ -4,24 +4,37 @@ import bluetooth
from ble_advertising import advertising_payload
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_UART_UUID = bluetooth.UUID('6E400001-B5A3-F393-E0A9-E50E24DCCA9E')
_UART_TX = (bluetooth.UUID('6E400003-B5A3-F393-E0A9-E50E24DCCA9E'), bluetooth.FLAG_NOTIFY,)
_UART_RX = (bluetooth.UUID('6E400002-B5A3-F393-E0A9-E50E24DCCA9E'), bluetooth.FLAG_WRITE,)
_UART_SERVICE = (_UART_UUID, (_UART_TX, _UART_RX,),)
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
_UART_TX = (
bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"),
bluetooth.FLAG_NOTIFY,
)
_UART_RX = (
bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"),
bluetooth.FLAG_WRITE,
)
_UART_SERVICE = (
_UART_UUID,
(_UART_TX, _UART_RX,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_COMPUTER = const(128)
class BLEUART:
def __init__(self, ble, name='mpy-uart', rxbuf=100):
def __init__(self, ble, name="mpy-uart", rxbuf=100):
self._ble = ble
self._ble.active(True)
self._ble.irq(handler=self._irq)
((self._tx_handle, self._rx_handle,),) = self._ble.gatts_register_services((_UART_SERVICE,))
((self._tx_handle, self._rx_handle,),) = self._ble.gatts_register_services(
(_UART_SERVICE,)
)
# Increase the size of the rx buffer and enable append mode.
self._ble.gatts_set_buffer(self._rx_handle, rxbuf, True)
self._connections = set()
@@ -82,7 +95,7 @@ def demo():
uart = BLEUART(ble)
def on_rx():
print('rx: ', uart.read().decode().strip())
print("rx: ", uart.read().decode().strip())
uart.irq(handler=on_rx)
nums = [4, 8, 15, 16, 23, 42]
@@ -90,7 +103,7 @@ def demo():
try:
while True:
uart.write(str(nums[i]) + '\n')
uart.write(str(nums[i]) + "\n")
i = (i + 1) % len(nums)
time.sleep_ms(1000)
except KeyboardInterrupt:
@@ -99,5 +112,5 @@ def demo():
uart.close()
if __name__ == '__main__':
if __name__ == "__main__":
demo()

Some files were not shown because too many files have changed in this diff Show More