18069996051/mixly3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix: 修复MicroPython MixGoAI和Microbit下一些py异常同时格式化代码

Browse Source
This commit is contained in:
王立帮
2024-12-03 10:36:49 +08:00
parent e6c9a30bdc
commit ee04dadb66
71 changed files with 3265 additions and 2382 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
boards/default_src/micropython_k210_mixgoai/origin/build/lib/adxl345.py
View File

@@ -3,12 +3,13 @@ import time
import ustruct
DATA_FORMAT = 0x31
BW_RATE = 0x2c
POWER_CTL = 0x2d
INT_ENABLE = 0x2E
OFSX = 0x1e
OFSY =0x1f
OFSZ =0x20
BW_RATE = 0x2C
POWER_CTL = 0x2D
INT_ENABLE = 0x2E
OFSX = 0x1E
OFSY = 0x1F
OFSZ = 0x20
class ADXL345:
def __init__(self, i2c):
@@ -18,51 +19,51 @@ class ADXL345:
for s in slv:
buf = self.i2c.readfrom_mem(s, 0, 1)
print(buf)
if(buf[0] == 0xe5):
if buf[0] == 0xE5:
self.slvAddr = s
print('adxl345 found')
print("adxl345 found")
break
#self.writeByte(POWER_CTL,0x00) #sleep
#time.sleep(0.001)
#低电平中断输出,13位全分辨率,输出数据右对齐,16g量程
self.writeByte(DATA_FORMAT,0x2B)
#数据输出速度为100Hz
self.writeByte(BW_RATE,0x0A)
#不使用中断
self.writeByte(INT_ENABLE,0x00)
# self.writeByte(POWER_CTL,0x00) #sleep
# time.sleep(0.001)
# 低电平中断输出,13位全分辨率,输出数据右对齐,16g量程
self.writeByte(DATA_FORMAT, 0x2B)
# 数据输出速度为100Hz
self.writeByte(BW_RATE, 0x0A)
# 不使用中断
self.writeByte(INT_ENABLE, 0x00)
self.writeByte(OFSX,0x00)
self.writeByte(OFSY,0x00)
self.writeByte(OFSZ,0x00)
#链接使能,测量模式
self.writeByte(POWER_CTL,0x28)
self.writeByte(OFSX, 0x00)
self.writeByte(OFSY, 0x00)
self.writeByte(OFSZ, 0x00)
# 链接使能,测量模式
self.writeByte(POWER_CTL, 0x28)
time.sleep(1)
def readXYZ(self):
fmt = '<h' #little-endian
fmt = "<h" # little-endian
buf1 = self.readByte(0x32)
buf2 = self.readByte(0x33)
buf = bytearray([buf1[0], buf2[0]])
x, = ustruct.unpack(fmt, buf)
x = x*3.9
#print('x:',x)
(x,) = ustruct.unpack(fmt, buf)
x = x * 3.9
# print('x:',x)
buf1 = self.readByte(0x34)
buf2 = self.readByte(0x35)
buf = bytearray([buf1[0], buf2[0]])
y, = ustruct.unpack(fmt, buf)
y = y*3.9
#print('y:',y)
(y,) = ustruct.unpack(fmt, buf)
y = y * 3.9
# print('y:',y)
buf1 = self.readByte(0x36)
buf2 = self.readByte(0x37)
buf = bytearray([buf1[0], buf2[0]])
z, = ustruct.unpack(fmt, buf)
z = z*3.9
#print('z:',z)
#print('************************')
#time.sleep(0.5)
return (x,y,z)
(z,) = ustruct.unpack(fmt, buf)
z = z * 3.9
# print('z:',z)
# print('************************')
# time.sleep(0.5)
return (x, y, z)
def readX(self):
return self.readXYZ()[0]
@@ -76,17 +77,19 @@ class ADXL345:
def writeByte(self, addr, data):
d = bytearray([data])
self.i2c.writeto_mem(self.slvAddr, addr, d)
def readByte(self, addr):
return self.i2c.readfrom_mem(self.slvAddr, addr, 1)
if __name__ == '__main__':
#初始化
sensor = ADXL345(i2c)
#getXYZ
x,y,z = sensor.readXYZ()
if __name__ == "__main__":
# 初始化
sensor = ADXL345(i2c)
#getX/Y/Z
x0 = sensor.readX()
y0 = sensor.readY()
z0 = sensor.readZ()
# getXYZ
x, y, z = sensor.readXYZ()
# getX/Y/Z
x0 = sensor.readX()
y0 = sensor.readY()
z0 = sensor.readZ()

227
boards/default_src/micropython_k210_mixgoai/origin/build/lib/ai_face.py
View File

@@ -1,16 +1,16 @@
import KPU as kpu
import gc,image,time
import gc, image, time
import board
try:
kpu.deinit(task_fe)
kpu.deinit(task_ld)
kpu.deinit(task_fd)
del task_fe
del task_ld
del task_fd
kpu.deinit(task_fe)
kpu.deinit(task_ld)
kpu.deinit(task_fd)
del task_fe
del task_ld
del task_fd
except Exception:
pass
@@ -20,114 +20,129 @@ record_ftr = []
record_ftrs = []
img_face = image.Image(size=(128, 128))
a = img_face.pix_to_ai()
dst_point = [(44, 59), (84, 59), (64, 82), (47, 105),(81, 105)]
dst_point = [(44, 59), (84, 59), (64, 82), (47, 105), (81, 105)]
start_processing = False
tim2 = time.ticks_ms()
task_fd=None
task_ld=None
task_fe=None
info=None
bb=1
task_fd = None
task_ld = None
task_fe = None
info = None
bb = 1
def set_key_state(*_):
global start_processing
global tim2
if (time.ticks_ms() - tim2 )> 4000:
start_processing = True
tim2 = time.ticks_ms()
global start_processing
global tim2
if (time.ticks_ms() - tim2) > 4000:
start_processing = True
tim2 = time.ticks_ms()
def init(FD,LD,FE):
global task_fd
global task_ld
global task_fe
# task_fd = kpu.load(0x200000)
# task_ld = kpu.load(0x300000)
# task_fe = kpu.load(0x400000)
task_fd = kpu.load(FD)
task_ld = kpu.load(LD)
task_fe = kpu.load(FE)
def init(FD, LD, FE):
global task_fd
global task_ld
global task_fe
# task_fd = kpu.load(0x200000)
# task_ld = kpu.load(0x300000)
# task_fe = kpu.load(0x400000)
task_fd = kpu.load(FD)
task_ld = kpu.load(LD)
task_fe = kpu.load(FE)
gc.collect()
key_gpio = board.pin(9, board.GPIO.IN, board.GPIO.PULL_UP)
key_gpio.irq(set_key_state, board.GPIO.IRQ_RISING, board.GPIO.WAKEUP_NOT_SUPPORT)
anchor = (
1.889,
2.5245,
2.9465,
3.94056,
3.99987,
5.3658,
5.155437,
6.92275,
6.718375,
9.01025,
) # anchor for face detect
kpu.init_yolo2(task_fd, 0.5, 0.3, 5, anchor)
def train(img, names, threshold):
global task_fd
global task_ld
global task_fe
global start_processing
global info
global bb
code = kpu.run_yolo2(task_fd, img)
if code:
for i in code:
face_cut = img.cut(i.x(), i.y(), i.w(), i.h())
face_cut_128 = face_cut.resize(128, 128)
a = face_cut_128.pix_to_ai()
fmap = kpu.forward(task_ld, face_cut_128)
plist = fmap[:]
le = (i.x() + int(plist[0] * i.w() - 10), i.y() + int(plist[1] * i.h()))
re = (i.x() + int(plist[2] * i.w()), i.y() + int(plist[3] * i.h()))
nose = (i.x() + int(plist[4] * i.w()), i.y() + int(plist[5] * i.h()))
lm = (i.x() + int(plist[6] * i.w()), i.y() + int(plist[7] * i.h()))
rm = (i.x() + int(plist[8] * i.w()), i.y() + int(plist[9] * i.h()))
lb = i.rect()
src_point = [le, re, nose, lm, rm]
T = image.get_affine_transform(src_point, dst_point)
a = image.warp_affine_ai(img, img_face, T)
a = img_face.ai_to_pix()
del face_cut_128
fmap = kpu.forward(task_fe, img_face)
feature = kpu.face_encode(fmap[:])
reg_flag = False
scores = []
for j in range(len(record_ftrs)):
score = kpu.face_compare(record_ftrs[j], feature)
scores.append(score)
max_score = 0
index = 0
for k in range(len(scores)):
if max_score < scores[k]:
max_score = scores[k]
index = k
if start_processing:
record_ftr = feature
record_ftrs.append(record_ftr)
start_processing = False
if max_score > threshold:
info = [names[index], max_score, lb, src_point]
else:
if bb == 1:
print("Please press BOOT key to enter the face")
bb = 0
info = [None, max_score, lb, src_point]
return True
break
else:
info = None
bb = 1
return False
gc.collect()
gc.collect()
key_gpio = board.pin(9,board.GPIO.IN,board.GPIO.PULL_UP)
key_gpio.irq(set_key_state,board.GPIO.IRQ_RISING, board.GPIO.WAKEUP_NOT_SUPPORT)
anchor = (1.889, 2.5245, 2.9465, 3.94056, 3.99987, 5.3658, 5.155437,6.92275, 6.718375, 9.01025) # anchor for face detect
kpu.init_yolo2(task_fd, 0.5, 0.3, 5, anchor)
def train(img,names,threshold):
global task_fd
global task_ld
global task_fe
global start_processing
global info
global bb
code = kpu.run_yolo2(task_fd, img)
if code:
for i in code:
face_cut = img.cut(i.x(), i.y(), i.w(), i.h())
face_cut_128 = face_cut.resize(128, 128)
a = face_cut_128.pix_to_ai()
fmap = kpu.forward(task_ld, face_cut_128)
plist = fmap[:]
le = (i.x()+int(plist[0]*i.w() - 10), i.y()+int(plist[1]*i.h()))
re = (i.x()+int(plist[2]*i.w()), i.y()+int(plist[3]*i.h()))
nose = (i.x()+int(plist[4]*i.w()), i.y()+int(plist[5]*i.h()))
lm = (i.x()+int(plist[6]*i.w()), i.y()+int(plist[7]*i.h()))
rm = (i.x()+int(plist[8]*i.w()), i.y()+int(plist[9]*i.h()))
lb=i.rect()
src_point = [le, re, nose, lm, rm]
T = image.get_affine_transform(src_point, dst_point)
a = image.warp_affine_ai(img, img_face, T)
a = img_face.ai_to_pix()
del(face_cut_128)
fmap = kpu.forward(task_fe, img_face)
feature = kpu.face_encode(fmap[:])
reg_flag = False
scores = []
for j in range(len(record_ftrs)):
score = kpu.face_compare(record_ftrs[j], feature)
scores.append(score)
max_score = 0
index = 0
for k in range(len(scores)):
if max_score < scores[k]:
max_score = scores[k]
index = k
if start_processing:
record_ftr = feature
record_ftrs.append(record_ftr)
start_processing = False
if max_score > threshold:
info=[names[index],max_score,lb,src_point]
else:
if bb==1:
print("Please press BOOT key to enter the face")
bb=0
info=[None,max_score,lb,src_point]
return True
break
else:
info=None
bb=1
return False
gc.collect()
def info_name():
gc.collect()
return info[0]
gc.collect()
return info[0]
def info_score():
return info[1]
def info_face():
return info[2]
def info_organs():
return info[3]
def info_score():
return info[1]
def info_face():
return info[2]
def info_organs():
return info[3]

121
boards/default_src/micropython_k210_mixgoai/origin/build/lib/aionenet.py
View File

@@ -1,25 +1,29 @@
import network,time,random,request,base64,json,board
import network, time, random, request, base64, json, board
from machine import UART
wifi_en=board.pin(19,board.GPIO.OUT)
board.register(18,board.FPIOA.UART2_TX)
board.register(17,board.FPIOA.UART2_RX)
wifi_en = board.pin(19, board.GPIO.OUT)
board.register(18, board.FPIOA.UART2_TX)
board.register(17, board.FPIOA.UART2_RX)
def wifi_enable(en):
global wifi_en
wifi_en.value(en)
def wifi_reset():
global uart
wifi_enable(0)
time.sleep_ms(200)
wifi_enable(1)
time.sleep(2)
uart = UART(UART.UART2,115200,timeout=1000, read_buf_len=4096)
uart = UART(UART.UART2, 115200, timeout=1000, read_buf_len=4096)
tmp = uart.read()
uart.write("AT+UART_CUR=921600,8,1,0,0\r\n")
print(uart.read())
uart = UART(UART.UART2,921600,timeout=1000, read_buf_len=10240) # important! baudrate too low or read_buf_len too small will loose data
uart = UART(
UART.UART2, 921600, timeout=1000, read_buf_len=10240
) # important! baudrate too low or read_buf_len too small will loose data
uart.write("AT\r\n")
tmp = uart.read()
print(tmp)
@@ -32,16 +36,17 @@ def wifi_reset():
return None
return nic
def nic_init(account,password):
nic=wifi_reset()
if not nic:
raise Exception("[Cool.AI]:WiFi init fail")
nic.connect(account,password)
nic.ifconfig()
def nic_init(account, password):
nic = wifi_reset()
if not nic:
raise Exception("[Cool.AI]:WiFi init fail")
nic.connect(account, password)
nic.ifconfig()
class SimpleEncode():
class SimpleEncode:
keyStr = "ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890abcdefghijklmnopqrstuvwxyz~!@#$%^&*()_+-={}[]:;<,>.?/|"
keyLength = len(keyStr)
encryptionA = 17
@@ -51,87 +56,87 @@ class SimpleEncode():
postCount = 5
randomChar = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnop"
randomCharLength = len(randomChar)
#base64字符
# base64字符
ALPHABET = "ABCDEFGHIJKLMN0123456789OPQRSTUVWXYZ+/abcdefghijklmnopqrstuvwxyz"
STANDARD = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" # 标准的字符串索引
#找到密钥
for i in range(1,keyLength):
if (encryptionA * i) % keyLength == 1:
decodeA = i
# 找到密钥
for i in range(1, keyLength):
if (encryptionA * i) % keyLength == 1:
decodeA = i
def base64Encode(self,sourceStr):
def base64Encode(self, sourceStr):
encode = ""
for ch in base64.b64encode(sourceStr.encode()).decode():
if ch == '=':
encode += '='
if ch == "=":
encode += "="
else:
for i in range(64):
if ch == self.STANDARD[i]:
encode += self.ALPHABET[i]
return encode
def encrpyt(self,sourceStr):
def encrpyt(self, sourceStr):
srcLength = len(sourceStr)
#先加入干扰字符的数量
addCharCount = random.randint(1,self.preCountMax) if srcLength < self.preCountMax else 0
#随机字符
sb = str(addCharCount)+"|"
# 先加入干扰字符的数量
addCharCount = (
random.randint(1, self.preCountMax) if srcLength < self.preCountMax else 0
)
# 随机字符
sb = str(addCharCount) + "|"
for i in range(addCharCount):
sb += self.randomChar[random.randint(0,self.randomCharLength-1)]
sb += self.randomChar[random.randint(0, self.randomCharLength - 1)]
sb += sourceStr
#尾部固定增加x个字符
# 尾部固定增加x个字符
for i in range(self.postCount):
sb += self.randomChar[random.randint(0,self.randomCharLength-1)]
sb += self.randomChar[random.randint(0, self.randomCharLength - 1)]
#base64 加密
# base64 加密
base64Str = self.base64Encode(sb)
destStr = ''
destStr = ""
for i in range(len(base64Str)):
#找到字符所在位置
# 找到字符所在位置
position = self.keyStr.find(base64Str[i])
#对字符进行转换
# 对字符进行转换
y = (self.encryptionA * position + self.encryptionB) % self.keyLength
#找到替换后的字符
# 找到替换后的字符
destStr += self.keyStr[y]
return destStr
def token(account,password):
url = 'http://ai.heclouds.com:9090/v1/user/oneNetLogin'
def token(account, password):
url = "http://ai.heclouds.com:9090/v1/user/oneNetLogin"
headers = {
'Content-Type': 'application/json',
"Content-Type": "application/json",
}
simpleencode = SimpleEncode()
data = {'account': account, 'password': simpleencode.encrpyt(password)}
data = {"account": account, "password": simpleencode.encrpyt(password)}
req = request.post(url, data=json.dumps(data), headers=headers)
#print(req.text)
# print(req.text)
try:
return eval(req.text)['data']['loginToken']
return eval(req.text)["data"]["loginToken"]
except:
raise ValueError( "[Cool.AI]:Wrong account or password ")
raise ValueError("[Cool.AI]:Wrong account or password ")
def post_ai(img,urlx,Token):
ur = 'http://183.230.40.32:9090/v1/aiApi/picture/MixPY'
url=ur.replace("MixPY",urlx)
headers ={
'Content-Type':'application/json',
'Login-Token':'Token'
}
headers['Login-Token']=Token
imge = img.compressed(quality=50)
file =imge.to_bytes()
str = base64.b64encode(file).decode()
data = {'picture':[str]}
req = request.post(url,data=json.dumps(data),headers=headers)
return json.loads(req.text)
#return eval(req.text)
def post_ai(img, urlx, Token):
ur = "http://183.230.40.32:9090/v1/aiApi/picture/MixPY"
url = ur.replace("MixPY", urlx)
headers = {"Content-Type": "application/json", "Login-Token": "Token"}
headers["Login-Token"] = Token
imge = img.compressed(quality=50)
file = imge.to_bytes()
str = base64.b64encode(file).decode()
data = {"picture": [str]}
req = request.post(url, data=json.dumps(data), headers=headers)
return json.loads(req.text)
# return eval(req.text)

205
boards/default_src/micropython_k210_mixgoai/origin/build/lib/base64.py
View File

@@ -3,25 +3,38 @@ import struct
import binascii
__all__ = [
'encode', 'decode', 'encodebytes', 'decodebytes',
'b64encode', 'b64decode', 'b32encode', 'b32decode',
'b16encode', 'b16decode',
'standard_b64encode', 'standard_b64decode',
'urlsafe_b64encode', 'urlsafe_b64decode',
]
"encode",
"decode",
"encodebytes",
"decodebytes",
"b64encode",
"b64decode",
"b32encode",
"b32decode",
"b16encode",
"b16decode",
"standard_b64encode",
"standard_b64decode",
"urlsafe_b64encode",
"urlsafe_b64decode",
]
bytes_types = (bytes, bytearray)
bytes_types = (bytes, bytearray)
def _bytes_from_decode_data(s):
if isinstance(s, str):
try:
return s.encode('ascii')
return s.encode("ascii")
except:
raise ValueError('string argument should contain only ASCII characters')
raise ValueError("string argument should contain only ASCII characters")
elif isinstance(s, bytes_types):
return s
else:
raise TypeError("argument should be bytes or ASCII string, not %s" % s.__class__.__name__)
raise TypeError(
"argument should be bytes or ASCII string, not %s" % s.__class__.__name__
)
def b64encode(s, altchars=None):
if not isinstance(s, bytes_types):
@@ -29,45 +42,73 @@ def b64encode(s, altchars=None):
encoded = binascii.b2a_base64(s)[:-1]
if altchars is not None:
if not isinstance(altchars, bytes_types):
raise TypeError("expected bytes, not %s"
% altchars.__class__.__name__)
raise TypeError("expected bytes, not %s" % altchars.__class__.__name__)
assert len(altchars) == 2, repr(altchars)
return encoded.translate(bytes.maketrans(b'+/', altchars))
return encoded.translate(bytes.maketrans(b"+/", altchars))
return encoded
def b64decode(s, altchars=None, validate=False):
s = _bytes_from_decode_data(s)
if altchars is not None:
altchars = _bytes_from_decode_data(altchars)
assert len(altchars) == 2, repr(altchars)
s = s.translate(bytes.maketrans(altchars, b'+/'))
if validate and not re.match(b'^[A-Za-z0-9+/]*={0,2}$', s):
raise binascii.Error('Non-base64 digit found')
s = s.translate(bytes.maketrans(altchars, b"+/"))
if validate and not re.match(b"^[A-Za-z0-9+/]*={0,2}$", s):
raise binascii.Error("Non-base64 digit found")
return binascii.a2b_base64(s)
def standard_b64encode(s):
return b64encode(s)
def standard_b64decode(s):
return b64decode(s)
def urlsafe_b64encode(s):
raise NotImplementedError()
def urlsafe_b64decode(s):
raise NotImplementedError()
_b32alphabet = {
0: b'A', 9: b'J', 18: b'S', 27: b'3',
1: b'B', 10: b'K', 19: b'T', 28: b'4',
2: b'C', 11: b'L', 20: b'U', 29: b'5',
3: b'D', 12: b'M', 21: b'V', 30: b'6',
4: b'E', 13: b'N', 22: b'W', 31: b'7',
5: b'F', 14: b'O', 23: b'X',
6: b'G', 15: b'P', 24: b'Y',
7: b'H', 16: b'Q', 25: b'Z',
8: b'I', 17: b'R', 26: b'2',
}
0: b"A",
9: b"J",
18: b"S",
27: b"3",
1: b"B",
10: b"K",
19: b"T",
28: b"4",
2: b"C",
11: b"L",
20: b"U",
29: b"5",
3: b"D",
12: b"M",
21: b"V",
30: b"6",
4: b"E",
13: b"N",
22: b"W",
31: b"7",
5: b"F",
14: b"O",
23: b"X",
6: b"G",
15: b"P",
24: b"Y",
7: b"H",
16: b"Q",
25: b"Z",
8: b"I",
17: b"R",
26: b"2",
}
_b32tab = [v[0] for k, v in sorted(_b32alphabet.items())]
_b32rev = dict([(v[0], k) for k, v in _b32alphabet.items()])
@@ -82,26 +123,29 @@ def b32encode(s):
quanta += 1
encoded = bytearray()
for i in range(quanta):
c1, c2, c3 = struct.unpack('!HHB', s[i*5:(i+1)*5])
c2 += (c1 & 1) << 16
c3 += (c2 & 3) << 8
encoded += bytes([_b32tab[c1 >> 11],
_b32tab[(c1 >> 6) & 0x1f],
_b32tab[(c1 >> 1) & 0x1f],
_b32tab[c2 >> 12],
_b32tab[(c2 >> 7) & 0x1f],
_b32tab[(c2 >> 2) & 0x1f],
_b32tab[c3 >> 5],
_b32tab[c3 & 0x1f],
])
c1, c2, c3 = struct.unpack("!HHB", s[i * 5 : (i + 1) * 5])
c2 += (c1 & 1) << 16
c3 += (c2 & 3) << 8
encoded += bytes(
[
_b32tab[c1 >> 11],
_b32tab[(c1 >> 6) & 0x1F],
_b32tab[(c1 >> 1) & 0x1F],
_b32tab[c2 >> 12],
_b32tab[(c2 >> 7) & 0x1F],
_b32tab[(c2 >> 2) & 0x1F],
_b32tab[c3 >> 5],
_b32tab[c3 & 0x1F],
]
)
if leftover == 1:
encoded = encoded[:-6] + b'======'
encoded = encoded[:-6] + b"======"
elif leftover == 2:
encoded = encoded[:-4] + b'===='
encoded = encoded[:-4] + b"===="
elif leftover == 3:
encoded = encoded[:-3] + b'==='
encoded = encoded[:-3] + b"==="
elif leftover == 4:
encoded = encoded[:-1] + b'='
encoded = encoded[:-1] + b"="
return bytes(encoded)
@@ -109,14 +153,14 @@ def b32decode(s, casefold=False, map01=None):
s = _bytes_from_decode_data(s)
quanta, leftover = divmod(len(s), 8)
if leftover:
raise binascii.Error('Incorrect padding')
raise binascii.Error("Incorrect padding")
if map01 is not None:
map01 = _bytes_from_decode_data(map01)
assert len(map01) == 1, repr(map01)
s = s.translate(bytes.maketrans(b'01', b'O' + map01))
s = s.translate(bytes.maketrans(b"01", b"O" + map01))
if casefold:
s = s.upper()
padchars = s.find(b'=')
padchars = s.find(b"=")
if padchars > 0:
padchars = len(s) - padchars
s = s[:-padchars]
@@ -129,16 +173,16 @@ def b32decode(s, casefold=False, map01=None):
for c in s:
val = _b32rev.get(c)
if val is None:
raise binascii.Error('Non-base32 digit found')
raise binascii.Error("Non-base32 digit found")
acc += _b32rev[c] << shift
shift -= 5
if shift < 0:
parts.append(binascii.unhexlify(bytes('%010x' % acc, "ascii")))
parts.append(binascii.unhexlify(bytes("%010x" % acc, "ascii")))
acc = 0
shift = 35
last = binascii.unhexlify(bytes('%010x' % acc, "ascii"))
last = binascii.unhexlify(bytes("%010x" % acc, "ascii"))
if padchars == 0:
last = b''
last = b""
elif padchars == 1:
last = last[:-1]
elif padchars == 3:
@@ -148,9 +192,10 @@ def b32decode(s, casefold=False, map01=None):
elif padchars == 6:
last = last[:-4]
else:
raise binascii.Error('Incorrect padding')
raise binascii.Error("Incorrect padding")
parts.append(last)
return b''.join(parts)
return b"".join(parts)
def b16encode(s):
if not isinstance(s, bytes_types):
@@ -162,12 +207,14 @@ def b16decode(s, casefold=False):
s = _bytes_from_decode_data(s)
if casefold:
s = s.upper()
if re.search(b'[^0-9A-F]', s):
raise binascii.Error('Non-base16 digit found')
if re.search(b"[^0-9A-F]", s):
raise binascii.Error("Non-base16 digit found")
return binascii.unhexlify(s)
MAXLINESIZE = 76
MAXBINSIZE = (MAXLINESIZE//4)*3
MAXLINESIZE = 76
MAXBINSIZE = (MAXLINESIZE // 4) * 3
def encode(input, output):
while True:
@@ -175,7 +222,7 @@ def encode(input, output):
if not s:
break
while len(s) < MAXBINSIZE:
ns = input.read(MAXBINSIZE-len(s))
ns = input.read(MAXBINSIZE - len(s))
if not ns:
break
s += ns
@@ -191,6 +238,7 @@ def decode(input, output):
s = binascii.a2b_base64(line)
output.write(s)
def encodebytes(s):
if not isinstance(s, bytes_types):
raise TypeError("expected bytes, not %s" % s.__class__.__name__)
@@ -200,10 +248,13 @@ def encodebytes(s):
pieces.append(binascii.b2a_base64(chunk))
return b"".join(pieces)
def encodestring(s):
import warnings
warnings.warn("encodestring() is a deprecated alias, use encodebytes()",
DeprecationWarning, 2)
warnings.warn(
"encodestring() is a deprecated alias, use encodebytes()", DeprecationWarning, 2
)
return encodebytes(s)
@@ -212,36 +263,50 @@ def decodebytes(s):
raise TypeError("expected bytes, not %s" % s.__class__.__name__)
return binascii.a2b_base64(s)
def decodestring(s):
import warnings
warnings.warn("decodestring() is a deprecated alias, use decodebytes()",
DeprecationWarning, 2)
warnings.warn(
"decodestring() is a deprecated alias, use decodebytes()", DeprecationWarning, 2
)
return decodebytes(s)
def main():
import sys, getopt
try:
opts, args = getopt.getopt(sys.argv[1:], 'deut')
opts, args = getopt.getopt(sys.argv[1:], "deut")
except getopt.error as msg:
sys.stdout = sys.stderr
print(msg)
print("""usage: %s [-d|-e|-u|-t] [file|-]
print(
"""usage: %s [-d|-e|-u|-t] [file|-]
-d, -u: decode
-e: encode (default)
-t: encode and decode string 'Aladdin:open sesame'"""%sys.argv[0])
-t: encode and decode string 'Aladdin:open sesame'"""
% sys.argv[0]
)
sys.exit(2)
func = encode
for o, a in opts:
if o == '-e': func = encode
if o == '-d': func = decode
if o == '-u': func = decode
if o == '-t': test(); return
if args and args[0] != '-':
with open(args[0], 'rb') as f:
if o == "-e":
func = encode
if o == "-d":
func = decode
if o == "-u":
func = decode
if o == "-t":
test()
return
if args and args[0] != "-":
with open(args[0], "rb") as f:
func(f, sys.stdout.buffer)
else:
func(sys.stdin.buffer, sys.stdout.buffer)
def test():
s0 = b"Aladdin:open sesame"
print(repr(s0))
@@ -252,5 +317,5 @@ def test():
assert s0 == s2
if __name__ == '__main__':
if __name__ == "__main__":
main()

41
boards/default_src/micropython_k210_mixgoai/origin/build/lib/bmp280.py
View File

@@ -1,6 +1,7 @@
from ustruct import unpack as unp
import utime
from machine import I2C
# Author David Wahlund david@dafnet.se
# Power Modes
@@ -49,21 +50,20 @@ class BMP280:
self._bmp_i2c = i2c_bus
self._i2c_addr = addr
self.chip_id = self._read(BMP280_REGISTER_ID, 2)
self._T1 = unp('<H', self._read(BMP280_REGISTER_DIG_T1, 2))[0]
self._T2 = unp('<h', self._read(BMP280_REGISTER_DIG_T2, 2))[0]
self._T3 = unp('<h', self._read(BMP280_REGISTER_DIG_T3, 2))[0]
self._P1 = unp('<H', self._read(BMP280_REGISTER_DIG_P1, 2))[0]
self._P2 = unp('<h', self._read(BMP280_REGISTER_DIG_P2, 2))[0]
self._P3 = unp('<h', self._read(BMP280_REGISTER_DIG_P3, 2))[0]
self._P4 = unp('<h', self._read(BMP280_REGISTER_DIG_P4, 2))[0]
self._P5 = unp('<h', self._read(BMP280_REGISTER_DIG_P5, 2))[0]
self._P6 = unp('<h', self._read(BMP280_REGISTER_DIG_P6, 2))[0]
self._P7 = unp('<h', self._read(BMP280_REGISTER_DIG_P7, 2))[0]
self._P8 = unp('<h', self._read(BMP280_REGISTER_DIG_P8, 2))[0]
self._P9 = unp('<h', self._read(BMP280_REGISTER_DIG_P9, 2))[0]
self._T1 = unp("<H", self._read(BMP280_REGISTER_DIG_T1, 2))[0]
self._T2 = unp("<h", self._read(BMP280_REGISTER_DIG_T2, 2))[0]
self._T3 = unp("<h", self._read(BMP280_REGISTER_DIG_T3, 2))[0]
self._P1 = unp("<H", self._read(BMP280_REGISTER_DIG_P1, 2))[0]
self._P2 = unp("<h", self._read(BMP280_REGISTER_DIG_P2, 2))[0]
self._P3 = unp("<h", self._read(BMP280_REGISTER_DIG_P3, 2))[0]
self._P4 = unp("<h", self._read(BMP280_REGISTER_DIG_P4, 2))[0]
self._P5 = unp("<h", self._read(BMP280_REGISTER_DIG_P5, 2))[0]
self._P6 = unp("<h", self._read(BMP280_REGISTER_DIG_P6, 2))[0]
self._P7 = unp("<h", self._read(BMP280_REGISTER_DIG_P7, 2))[0]
self._P8 = unp("<h", self._read(BMP280_REGISTER_DIG_P8, 2))[0]
self._P9 = unp("<h", self._read(BMP280_REGISTER_DIG_P9, 2))[0]
self._t_os = BMP280_TEMP_OS_2 # temperature oversampling
self._p_os = BMP280_PRES_OS_16 # pressure oversampling
@@ -96,7 +96,9 @@ class BMP280:
r = self._t_os + (self._p_os << 3) + (1 << 6)
self._write(BMP280_REGISTER_CONTROL, r)
utime.sleep_ms(100) # TODO calc sleep
d = self._read(BMP280_REGISTER_DATA, 6) # read all data at once (as by spec)
d = self._read(
BMP280_REGISTER_DATA, 6
) # read all data at once (as by spec)
self._p_raw = (d[0] << 12) + (d[1] << 4) + (d[2] >> 4)
self._t_raw = (d[3] << 12) + (d[4] << 4) + (d[5] >> 4)
@@ -142,14 +144,20 @@ class BMP280:
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
var2 = (
(
(((self._t_raw >> 4) - self._T1) * ((self._t_raw >> 4) - self._T1))
>> 12
)
* self._T3
) >> 14
self._t_fine = var1 + var2
# @property
def get_BMP_temperature(self):
self._calc_t_fine()
if self._t == 0:
self._t = ((self._t_fine * 5 + 128) >> 8) / 100.
self._t = ((self._t_fine * 5 + 128) >> 8) / 100.0
return self._t
# @property
@@ -175,4 +183,3 @@ class BMP280:
p = ((p + var1 + var2) >> 8) + (self._P7 << 4)
self._p = p / 256.0
return self._p

40
boards/default_src/micropython_k210_mixgoai/origin/build/lib/ce_com.py
View File

@@ -6,24 +6,26 @@ MicroPython library for the MxiGo AI
20211213
mixly
"""
data_a=None
def uart_tx(uart,data,repeat=True):
global data_a
data_b = data
if data_b != data_a:
uart.write((str(data)+'\n'))
#print(data)
if not repeat:
data_a=data_b
data_a = None
def uart_tx(uart, data, repeat=True):
global data_a
data_b = data
if data_b != data_a:
uart.write((str(data) + "\n"))
# print(data)
if not repeat:
data_a = data_b
def uart_rx(uart):
data = uart.readline()
if data:
data_str = data.strip()
try:
data_str=data_str.decode()
return eval(data_str)
except:
return data_str
data = uart.readline()
if data:
data_str = data.strip()
try:
data_str = data_str.decode()
return eval(data_str)
except:
return data_str

116
boards/default_src/micropython_k210_mixgoai/origin/build/lib/dht11.py
View File

@@ -1,58 +1,60 @@
import time,board
import time, board
def read_data(pin_name):
data=[]
j=0
time.sleep_ms(1200)
N1 = board.pin(pin_name, board.GPIO.OUT)
N1.value(0)
time.sleep_ms(20)
N1.value(1)
time.sleep_us(30)
N1 =board.pin(pin_name, board.GPIO.IN)
T1 = time.ticks_us()
while N1.value()==0:
continue
while N1.value()==1:
T2 =time.ticks_us()
if time.ticks_diff(T2, T1) >200000:
#raise ValueError("[MixNo]:Sensor read error")
break
continue
while j<40:
k=0
while N1.value()==0:
continue
while N1.value()==1:
k+=1
if k>100:break
if k<15:
data.append(0)
else:
data.append(1)
j=j+1
del N1
humidity_bit=data[0:8]
humidity_point_bit=data[8:16]
temperature_bit=data[16:24]
temperature_point_bit=data[24:32]
check_bit=data[32:40]
humidity=0
humidity_point=0
temperature=0.0
temperature_point=0
check=0
for i in range(8):
humidity+=humidity_bit[i]*2**(7-i)
humidity_point+=humidity_point_bit[i]*2**(7-i)
temperature+=temperature_bit[i]*2**(7-i)
temperature_point+=temperature_point_bit[i]*2**(7-i)
check+=check_bit[i]*2**(7-i)
tmp=humidity+humidity_point+temperature+temperature_point
#print(humidity_point,temperature_point)
if check==tmp:
#print('temperature is',temperature,'-wet is',humidity,'%')
return (temperature+temperature_point/10,humidity)
else:
#print('Error:',humidity,humidity_point,temperature,temperature_point,check)
return (None,None)
data = []
j = 0
time.sleep_ms(1200)
N1 = board.pin(pin_name, board.GPIO.OUT)
N1.value(0)
time.sleep_ms(20)
N1.value(1)
time.sleep_us(30)
N1 = board.pin(pin_name, board.GPIO.IN)
T1 = time.ticks_us()
while N1.value() == 0:
continue
while N1.value() == 1:
T2 = time.ticks_us()
if time.ticks_diff(T2, T1) > 200000:
# raise ValueError("[MixNo]:Sensor read error")
break
continue
while j < 40:
k = 0
while N1.value() == 0:
continue
while N1.value() == 1:
k += 1
if k > 100:
break
if k < 15:
data.append(0)
else:
data.append(1)
j = j + 1
del N1
humidity_bit = data[0:8]
humidity_point_bit = data[8:16]
temperature_bit = data[16:24]
temperature_point_bit = data[24:32]
check_bit = data[32:40]
humidity = 0
humidity_point = 0
temperature = 0.0
temperature_point = 0
check = 0
for i in range(8):
humidity += humidity_bit[i] * 2 ** (7 - i)
humidity_point += humidity_point_bit[i] * 2 ** (7 - i)
temperature += temperature_bit[i] * 2 ** (7 - i)
temperature_point += temperature_point_bit[i] * 2 ** (7 - i)
check += check_bit[i] * 2 ** (7 - i)
tmp = humidity + humidity_point + temperature + temperature_point
# print(humidity_point,temperature_point)
if check == tmp:
# print('temperature is',temperature,'-wet is',humidity,'%')
return (temperature + temperature_point / 10, humidity)
else:
# print('Error:',humidity,humidity_point,temperature,temperature_point,check)
return (None, None)

51
boards/default_src/micropython_k210_mixgoai/origin/build/lib/hcr04.py
View File

@@ -1,27 +1,28 @@
import time,board
import time, board
def Sonar(trig1, echo1):
trig = board.pin(trig1, board.GPIO.OUT)
echo = board.pin(echo1, board.GPIO.IN)
time.sleep_ms(10)
trig.value(1)
time.sleep_us(10)
trig.value(0)
n1 = time.ticks_us()
while(echo.value()==0):
n2 =time.ticks_us()
if time.ticks_diff(n2, n1) >200000:
#raise ValueError("[MixNo]:Sensor read error")
break
pass
t1 = time.ticks_us()
while(echo.value()==1):
n3 =time.ticks_us()
if time.ticks_diff(n3, t1) >200000:
#raise ValueError("[MixNo]:Sensor read error")
break
pass
t2 = time.ticks_us()
time.sleep_ms(10)
return round(time.ticks_diff(t2, t1) / 10000 * 340 / 2, 2)
trig = board.pin(trig1, board.GPIO.OUT)
echo = board.pin(echo1, board.GPIO.IN)
time.sleep_ms(10)
trig.value(1)
time.sleep_us(10)
trig.value(0)
n1 = time.ticks_us()
while echo.value() == 0:
n2 = time.ticks_us()
if time.ticks_diff(n2, n1) > 200000:
# raise ValueError("[MixNo]:Sensor read error")
break
pass
t1 = time.ticks_us()
while echo.value() == 1:
n3 = time.ticks_us()
if time.ticks_diff(n3, t1) > 200000:
# raise ValueError("[MixNo]:Sensor read error")
break
pass
t2 = time.ticks_us()
time.sleep_ms(10)
return round(time.ticks_diff(t2, t1) / 10000 * 340 / 2, 2)

77
boards/default_src/micropython_k210_mixgoai/origin/build/lib/irremote.py
View File

@@ -1,47 +1,48 @@
import time,board
import time, board
def read_id(pin):
L1 = board.pin(pin, board.GPIO.IN, board.GPIO.PULL_UP)
a = []
t1 = time.ticks_us()
while L1.value() == 1:
t2 = time.ticks_us()
if time.ticks_diff(t2, t1) >1000000:
#raise ValueError("[MixNo]:Sensor read error")
break
pass
L1 = board.pin(pin, board.GPIO.IN, board.GPIO.PULL_UP)
a = []
t1 = time.ticks_us()
while L1.value() == 1:
t2 = time.ticks_us()
if time.ticks_diff(t2, t1) > 1000000:
# raise ValueError("[MixNo]:Sensor read error")
break
pass
time.sleep_us(13560)
time.sleep_us(13560)
for i in range(1000):
v = L1.value()
a.append(v)
time.sleep_us(56)
for i in range(1000):
v = L1.value()
a.append(v)
time.sleep_us(56)
a_c = []
count = 0
a_c = []
count = 0
for i in a:
if i == 1:
count += 1
for i in a:
if i == 1:
count += 1
elif i == 0:
if count > 0 :
a_c.append(count)
count =0
elif i == 0:
if count > 0:
a_c.append(count)
count = 0
for i in range(len(a_c)):
if a_c[i] > 10:
a_c[i] = "1"
else:
a_c[i] = "0"
B1 = "".join(a_c)
B2 = B1[16:32]
#print(len(B1))
if len(B1)==32 or len(B1)==33 or len(B1)==46:
B3=int(B2,2)
return B3
else:
return None
for i in range(len(a_c)):
if a_c[i] > 10:
a_c[i] = "1"
else:
a_c[i] = "0"
B1 = "".join(a_c)
B2 = B1[16:32]
# print(len(B1))
if len(B1) == 32 or len(B1) == 33 or len(B1) == 46:
B3 = int(B2, 2)
return B3
else:
return None

221
boards/default_src/micropython_k210_mixgoai/origin/build/lib/map.json Normal file
View File

@@ -0,0 +1,221 @@
{
"adxl345": {
"__require__": [
"machine",
"time",
"ustruct"
],
"__file__": true,
"__name__": "adxl345.py"
},
"aionenet": {
"__require__": [
"network",
"time",
"random",
"request",
"base64",
"json",
"board",
"machine"
],
"__file__": true,
"__name__": "aionenet.py"
},
"ai_face": {
"__require__": [
"KPU",
"gc",
"image",
"time",
"board"
],
"__file__": true,
"__name__": "ai_face.py"
},
"base64": {
"__require__": [
"re",
"struct",
"binascii",
"warnings",
"warnings",
"sys",
"getopt"
],
"__file__": true,
"__name__": "base64.py"
},
"bmp280": {
"__require__": [
"ustruct",
"utime",
"machine"
],
"__file__": true,
"__name__": "bmp280.py"
},
"ce_com": {
"__require__": [],
"__file__": true,
"__name__": "ce_com.py"
},
"dht11": {
"__require__": [
"time",
"board"
],
"__file__": true,
"__name__": "dht11.py"
},
"hcr04": {
"__require__": [
"time",
"board"
],
"__file__": true,
"__name__": "hcr04.py"
},
"irremote": {
"__require__": [
"time",
"board"
],
"__file__": true,
"__name__": "irremote.py"
},
"matcher": {
"__require__": [],
"__file__": true,
"__name__": "matcher.py"
},
"miot_no": {
"__require__": [
"usocket",
"ustruct",
"network",
"time",
"board",
"ujson",
"machine",
"ussl"
],
"__file__": true,
"__name__": "miot_no.py"
},
"mixiot": {
"__require__": [
"usocket",
"ustruct",
"time",
"machine",
"ubinascii",
"ujson",
"matcher",
"machine",
"ussl"
],
"__file__": true,
"__name__": "mixiot.py"
},
"mixly_tool": {
"__require__": [
"hashlib",
"os",
"sys"
],
"__file__": true,
"__name__": "mixly_tool.py"
},
"mixpy": {
"__require__": [
"math"
],
"__file__": true,
"__name__": "mixpy.py"
},
"mpu9250": {
"__require__": [
"micropython",
"ustruct",
"utime",
"time",
"math",
"math",
"os",
"compass_cfg"
],
"__file__": true,
"__name__": "mpu9250.py"
},
"ms32006": {
"__require__": [
"time",
"micropython"
],
"__file__": true,
"__name__": "ms32006.py"
},
"net_espat": {
"__require__": [
"network",
"time",
"board",
"machine",
"time"
],
"__file__": true,
"__name__": "net_espat.py"
},
"pid": {
"__require__": [
"time",
"math"
],
"__file__": true,
"__name__": "pid.py"
},
"player": {
"__require__": [
"board",
"audio",
"video",
"Maix",
"gc",
"sensor",
"lcd",
"lcd"
],
"__file__": true,
"__name__": "player.py"
},
"reset": {
"__require__": [
"image",
"lcd",
"time",
"gc",
"machine",
"gc"
],
"__file__": true,
"__name__": "reset.py"
},
"servo": {
"__require__": [
"math",
"ustruct",
"time"
],
"__file__": true,
"__name__": "servo.py"
},
"sht20": {
"__require__": [
"struct",
"time"
],
"__file__": true,
"__name__": "sht20.py"
}
}

88
boards/default_src/micropython_k210_mixgoai/origin/build/lib/miot_no.py
View File

@@ -1,29 +1,34 @@
import usocket as socket
import ustruct as struct
import network,time,board
#from ubinascii import hexlify
import network, time, board
# from ubinascii import hexlify
import ujson as json
from machine import UART
wifi_en=board.pin(19,board.GPIO.OUT)
board.register(18,board.FPIOA.UART2_TX)
board.register(17,board.FPIOA.UART2_RX)
wifi_en = board.pin(19, board.GPIO.OUT)
board.register(18, board.FPIOA.UART2_TX)
board.register(17, board.FPIOA.UART2_RX)
def wifi_enable(en):
global wifi_en
wifi_en.value(en)
def wifi_reset():
global uart
wifi_enable(0)
time.sleep_ms(200)
wifi_enable(1)
time.sleep(2)
uart = UART(UART.UART2,115200,timeout=1000, read_buf_len=4096)
uart = UART(UART.UART2, 115200, timeout=1000, read_buf_len=4096)
tmp = uart.read()
uart.write("AT+UART_CUR=921600,8,1,0,0\r\n")
print(uart.read())
uart = UART(UART.UART2,921600,timeout=1000, read_buf_len=10240) # important! baudrate too low or read_buf_len too small will loose data
uart = UART(
UART.UART2, 921600, timeout=1000, read_buf_len=10240
) # important! baudrate too low or read_buf_len too small will loose data
uart.write("AT\r\n")
tmp = uart.read()
print(tmp)
@@ -36,46 +41,62 @@ def wifi_reset():
return None
return nic
def get_data_dict(d):
result = {"datastreams":[]}
result = {"datastreams": []}
for x in d:
result["datastreams"].append({"id":x,"datapoints":[{"value":d[x]}]})
result["datastreams"].append({"id": x, "datapoints": [{"value": d[x]}]})
return result
def pubData(value, state):
def pubData(value, state):
value = get_data_dict(value)
jdata = json.dumps(value)
jlen = len(jdata)
bdata = bytearray(jlen+3)
bdata[0] = 1 # publish data in type of json
bdata[1] = int(jlen / 256) # data lenght
bdata[2] = jlen % 256 # data lenght
bdata[3:jlen+4] = jdata.encode('ascii') # json data
bdata = bytearray(jlen + 3)
bdata[0] = 1 # publish data in type of json
bdata[1] = int(jlen / 256) # data lenght
bdata[2] = jlen % 256 # data lenght
bdata[3 : jlen + 4] = jdata.encode("ascii") # json data
if state:
print(value)
print(bdata)
return bdata
def do_connect(account,password):
nic=wifi_reset()
if not nic:
raise Exception("[Cool.AI]:WiFi init fail")
nic.connect(account,password)
nic.ifconfig()
def do_connect(account, password):
nic = wifi_reset()
if not nic:
raise Exception("[Cool.AI]:WiFi init fail")
nic.connect(account, password)
nic.ifconfig()
def init_MQTT_client(sid, address, cid, api, topic, callback):
client = MQTTClient(sid, address, 6002, cid, api)
client.set_callback(callback)
client.connect()
client.subscribe(bytes(topic, 'utf-8'))
client.subscribe(bytes(topic, "utf-8"))
return client
class MQTTException(Exception):
pass
class MQTTClient:
def __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0,ssl=False, ssl_params={}):
def __init__(
self,
client_id,
server,
port=0,
user=None,
password=None,
keepalive=0,
ssl=False,
ssl_params={},
):
if port == 0:
port = 8883 if ssl else 1883
self.client_id = client_id
@@ -102,7 +123,7 @@ class MQTTClient:
sh = 0
while 1:
b = self.sock.read(1)[0]
n |= (b & 0x7f) << sh
n |= (b & 0x7F) << sh
if not b & 0x80:
return n
sh += 7
@@ -124,6 +145,7 @@ class MQTTClient:
print(self.addr)
if self.ssl:
import ussl
self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)
msg = bytearray(b"\x10\0\0\x04MQTT\x04\x02\0\0")
msg[1] = 10 + 2 + len(self.client_id)
@@ -140,7 +162,7 @@ class MQTTClient:
msg[9] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3
msg[9] |= self.lw_retain << 5
self.sock.write(msg)
#print(hex(len(msg)), hexlify(msg, ":"))
# print(hex(len(msg)), hexlify(msg, ":"))
self._send_str(self.client_id)
if self.lw_topic:
self._send_str(self.lw_topic)
@@ -161,7 +183,7 @@ class MQTTClient:
def ping(self):
self.sock.write(b"\xc0\0")
def publish(self, msg, is_print=True, topic='$dp', retain=False, qos=0):
def publish(self, msg, is_print=True, topic="$dp", retain=False, qos=0):
msg = pubData(msg, is_print)
pkt = bytearray(b"\x30\0\0\0")
pkt[0] |= qos << 1 | retain
@@ -170,12 +192,12 @@ class MQTTClient:
sz += 2
assert sz < 2097152
i = 1
while sz > 0x7f:
pkt[i] = (sz & 0x7f) | 0x80
while sz > 0x7F:
pkt[i] = (sz & 0x7F) | 0x80
sz >>= 7
i += 1
pkt[i] = sz
#print(hex(len(pkt)), hexlify(pkt, ":"))
# print(hex(len(pkt)), hexlify(pkt, ":"))
self.sock.write(pkt, i + 1)
self._send_str(topic)
if qos > 0:
@@ -202,7 +224,7 @@ class MQTTClient:
pkt = bytearray(b"\x82\0\0\0")
self.pid += 1
struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)
#print(hex(len(pkt)), hexlify(pkt, ":"))
# print(hex(len(pkt)), hexlify(pkt, ":"))
self.sock.write(pkt)
self._send_str(topic)
self.sock.write(qos.to_bytes(1, "little"))
@@ -210,7 +232,7 @@ class MQTTClient:
op = self.wait_msg()
if op == 0x90:
resp = self.sock.read(4)
#print(resp)
# print(resp)
assert resp[1] == pkt[2] and resp[2] == pkt[3]
if resp[3] == 0x80:
raise MQTTException(resp[3])
@@ -227,12 +249,12 @@ class MQTTClient:
return None
if res == b"":
raise OSError(-1)
if res == b"\xd0": # PINGRESP
if res == b"\xd0": # PINGRESP
sz = self.sock.read(1)[0]
assert sz == 0
return None
op = res[0]
if op & 0xf0 != 0x30:
if op & 0xF0 != 0x30:
return op
sz = self._recv_len()
topic_len = self.sock.read(2)

106
boards/default_src/micropython_k210_mixgoai/origin/build/lib/mixiot.py
View File

@@ -7,33 +7,55 @@ import ujson as json
from matcher import MQTTMatcher
from machine import Timer
ADDITIONAL_TOPIC = 'b640a0ce465fa2a4150c36b305c1c11b'
WILL_TOPIC = '9d634e1a156dc0c1611eb4c3cff57276'
ADDITIONAL_TOPIC = "b640a0ce465fa2a4150c36b305c1c11b"
WILL_TOPIC = "9d634e1a156dc0c1611eb4c3cff57276"
def init_MQTT_client(address, username, password,MQTT_USR_PRJ):
def init_MQTT_client(address, username, password, MQTT_USR_PRJ):
client = MQTTClient(hexlify(machine.unique_id()), address, 1883, username, password)
client.set_last_will(topic=MQTT_USR_PRJ+WILL_TOPIC, msg=client.client_id, qos=2)
if client.connect()==0:
client.publish(MQTT_USR_PRJ+ADDITIONAL_TOPIC, client.client_id, qos=1)
Timer(Timer.TIMER2,Timer.CHANNEL3,mode=Timer.MODE_PERIODIC,period = 10000, callback = lambda x : client.ping())
client.set_last_will(topic=MQTT_USR_PRJ + WILL_TOPIC, msg=client.client_id, qos=2)
if client.connect() == 0:
client.publish(MQTT_USR_PRJ + ADDITIONAL_TOPIC, client.client_id, qos=1)
Timer(
Timer.TIMER2,
Timer.CHANNEL3,
mode=Timer.MODE_PERIODIC,
period=10000,
callback=lambda x: client.ping(),
)
return client
len_overrided = len
# Add by Mixly Team
def len(object):
if isinstance(object, str):
return len_overrided(object.encode('utf-8'))
return len_overrided(object.encode("utf-8"))
else:
return len_overrided(object)
#####################################################
#####################################################
class MQTTException(Exception):
pass
class MQTTClient:
def __init__(self, client_id, server, port=0, username=None, password=None, keepalive=60, ssl=False, ssl_params={}):
def __init__(
self,
client_id,
server,
port=0,
username=None,
password=None,
keepalive=60,
ssl=False,
ssl_params={},
):
if port == 0:
port = 8883 if ssl else 1883
self.client_id = client_id
@@ -42,11 +64,11 @@ class MQTTClient:
self.ssl = ssl
self.ssl_params = ssl_params
self.pid = 0
#self.cb = None
# self.cb = None
self._on_message = None
self.username = username
self.password = password
#self.project = project
# self.project = project
self.keepalive = keepalive
self.lw_topic = None
self.lw_msg = None
@@ -63,7 +85,7 @@ class MQTTClient:
sh = 0
while 1:
b = self.sock.read(1)[0]
n |= (b & 0x7f) << sh
n |= (b & 0x7F) << sh
if not b & 0x80:
return n
sh += 7
@@ -79,7 +101,7 @@ class MQTTClient:
"""
if mqtt_topic is None or callback_method is None:
raise ValueError("MQTT topic and callback method must both be defined.")
self._on_message_filtered[MQTT_USR_PRJ+mqtt_topic] = callback_method
self._on_message_filtered[MQTT_USR_PRJ + mqtt_topic] = callback_method
def remove_callback(self, mqtt_topic):
"""Removes a registered callback method.
@@ -93,7 +115,7 @@ class MQTTClient:
except KeyError:
raise KeyError(
"MQTT topic callback not added with add_topic_callback."
) from None
) from None
@property
def on_message(self):
@@ -124,19 +146,20 @@ class MQTTClient:
self.lw_msg = msg
self.lw_qos = qos
self.lw_retain = retain
def connect(self, clean_session=True):
self.sock = socket.socket()
self.sock.connect(self.addr)
print(self.addr)
if self.ssl:
import ussl
self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)
msg_header=bytearray([0x10])
msg_header = bytearray([0x10])
msg = bytearray(b"\x04MQTT\x04\x02\0\0")
msg_length = 12 + len(self.client_id)
msg[6] = clean_session << 1
if self.username is not None:
msg_length += 2 + len(self.username) + 2 + len(self.password)
msg[6] |= 0xC0
@@ -148,22 +171,22 @@ class MQTTClient:
msg_length += 2 + len(self.lw_topic) + 2 + len(self.lw_msg)
msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3
msg[6] |= self.lw_retain << 5
if msg_length > 0x7F:
while msg_length>0:
while msg_length > 0:
encoded_byte = msg_length % 0x80
msg_length = msg_length // 0x80
if msg_length > 0:
encoded_byte |= 0x80
msg_header.append(encoded_byte)
msg_header.append(encoded_byte)
msg_header.append(0x00)
else:
msg_header.append(msg_length)
msg_header.append(0x00)
self.sock.write(msg_header)
self.sock.write(msg_header)
self.sock.write(msg)
#print(hexlify(msg_header, ":"), hexlify(msg, ":"))
# print(hexlify(msg_header, ":"), hexlify(msg, ":"))
self._send_str(self.client_id)
if self.lw_topic:
self._send_str(self.lw_topic)
@@ -177,26 +200,25 @@ class MQTTClient:
raise MQTTException(resp[3])
return resp[2] & 1
def disconnect(self,MQTT_USR_PRJ):
#MQTT_USR_PRJ = "{}/{}/".format(self.username,self.project)
self.publish(MQTT_USR_PRJ+WILL_TOPIC, self.client_id, qos=1)
def disconnect(self, MQTT_USR_PRJ):
# MQTT_USR_PRJ = "{}/{}/".format(self.username,self.project)
self.publish(MQTT_USR_PRJ + WILL_TOPIC, self.client_id, qos=1)
self.sock.write(b"\xe0\0")
self.sock.close()
def ping(self):
self.sock.write(b"\xc0\0")
def pingSync(self):
time.ticks_ms()
self.ping()
for i in range(0,10):
for i in range(0, 10):
msg = self.check_msg()
if msg == "PINGRESP":
return True
time.sleep_ms(100)
return False
def publish(self, topic, msg, retain=False, qos=0):
# msg = pubData(msg)
if "+" in topic or "#" in topic:
@@ -219,12 +241,12 @@ class MQTTClient:
sz += 2
assert sz < 2097152
i = 1
while sz > 0x7f:
pkt[i] = (sz & 0x7f) | 0x80
while sz > 0x7F:
pkt[i] = (sz & 0x7F) | 0x80
sz >>= 7
i += 1
pkt[i] = sz
#print(hex(len(pkt)), hexlify(pkt, ":"))
# print(hex(len(pkt)), hexlify(pkt, ":"))
self.sock.write(pkt, i + 1)
self._send_str(topic)
if qos > 0:
@@ -247,13 +269,13 @@ class MQTTClient:
assert 0
def subscribe(self, topic, qos=0):
#assert self.cb is not None, "Subscribe callback is not set"
# assert self.cb is not None, "Subscribe callback is not set"
pkt = bytearray(b"\x82\0\0\0")
self.pid += 1
if isinstance(topic, str):
topic=topic.encode()
topic = topic.encode()
struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)
#print(hex(len(pkt)), hexlify(pkt, ":"))
# print(hex(len(pkt)), hexlify(pkt, ":"))
self.sock.write(pkt)
self._send_str(topic)
self.sock.write(qos.to_bytes(1, "little"))
@@ -261,7 +283,7 @@ class MQTTClient:
op = self.wait_msg()
if op == 0x90:
resp = self.sock.read(4)
#print(resp)
# print(resp)
assert resp[1] == pkt[2] and resp[2] == pkt[3]
if resp[3] == 0x80:
raise MQTTException(resp[3])
@@ -278,12 +300,12 @@ class MQTTClient:
return None
if res == b"":
raise OSError(-1)
if res == b"\xd0": # PINGRESP
if res == b"\xd0": # PINGRESP
sz = self.sock.read(1)[0]
assert sz == 0
return "PINGRESP"
op = res[0]
if op & 0xf0 != 0x30:
if op & 0xF0 != 0x30:
return op
sz = self._recv_len()
topic_len = self.sock.read(2)
@@ -296,7 +318,7 @@ class MQTTClient:
sz -= 2
msg = self.sock.read(sz)
self._handle_on_message(self, str(topic, "utf-8"), str(msg, "utf-8"))
#self.cb(topic.decode(), msg.decode())
# self.cb(topic.decode(), msg.decode())
if op & 6 == 2:
pkt = bytearray(b"\x40\x02\0\0")
struct.pack_into("!H", pkt, 2, pid)
@@ -309,4 +331,4 @@ class MQTTClient:
# the same processing as wait_msg.
def check_msg(self):
self.sock.setblocking(False)
return self.wait_msg()
return self.wait_msg()

10
boards/default_src/micropython_k210_mixgoai/origin/build/lib/mixly_tool.py
View File

@@ -1,25 +1,27 @@
import hashlib
import os
def sha_file(f):
if f not in set(os.listdir(".")):
return 'None:::sha_file_end'
return "None:::sha_file_end"
else:
sha = hashlib.sha256()
with open(f, encoding='utf-8') as fd:
with open(f, encoding="utf-8") as fd:
file_buffer = fd.read(128).encode("utf-8")
while len(file_buffer) > 0:
sha.update(file_buffer)
file_buffer = fd.read(128).encode("utf-8")
h = sha.digest()
return ''.join(['%.2x' % i for i in h]) + ":::sha_file_end"
return "".join(["%.2x" % i for i in h]) + ":::sha_file_end"
def reload(mod):
import sys
mod_name = mod.__name__
try:
del sys.modules[mod_name]
__import__(mod_name)
except:
pass

46
boards/default_src/micropython_k210_mixgoai/origin/build/lib/mixpy.py
View File

@@ -1,8 +1,9 @@
#coding=utf-8
# coding=utf-8
import math
def math_map(v, al, ah, bl, bh):
if al==ah:
if al == ah:
return bl
if al > ah:
al, ah = ah, al
@@ -12,19 +13,26 @@ def math_map(v, al, ah, bl, bh):
v = al
return bl + (bh - bl) * (v - al) / (ah - al)
def math_mean(myList):
localList = [e for e in myList if type(e) == int or type(e) == float]
if not localList: return
if not localList:
return
return float(sum(localList)) / len(localList)
def math_median(myList):
localList = sorted([e for e in myList if type(e) == int or type(e) == float])
if not localList: return
if not localList:
return
if len(localList) % 2 == 0:
return (localList[len(localList) // 2 - 1] + localList[len(localList) // 2]) / 2.0
return (
localList[len(localList) // 2 - 1] + localList[len(localList) // 2]
) / 2.0
else:
return localList[(len(localList) - 1) // 2]
def math_modes(some_list):
modes = []
# Using a lists of [item, count] to keep count rather than dict
@@ -45,40 +53,48 @@ def math_modes(some_list):
modes.append(counted_item)
return modes
def math_standard_deviation(numbers):
n = len(numbers)
if n == 0: return
if n == 0:
return
mean = float(sum(numbers)) / n
variance = sum((x - mean) ** 2 for x in numbers) / n
return math.sqrt(variance)
def lists_sort(my_list, type, reverse):
def try_float(s):
try:
return float(s)
except:
return 0
key_funcs = {
"NUMERIC": try_float,
"TEXT": str,
"IGNORE_CASE": lambda s: str(s).lower()
"IGNORE_CASE": lambda s: str(s).lower(),
}
key_func = key_funcs[type]
list_cpy = list(my_list)
return sorted(list_cpy, key=key_func, reverse=reverse)
def format_content(mydict, cid):
if 'lat' in mydict and 'long' in mydict:
res = '{'+'"lat": "{}", "long": "{}", "clientid": "{}"'.format(mydict.pop('lat'),mydict.pop('long'),cid)
if len(mydict)>0:
if "lat" in mydict and "long" in mydict:
res = "{" + '"lat": "{}", "long": "{}", "clientid": "{}"'.format(
mydict.pop("lat"), mydict.pop("long"), cid
)
if len(mydict) > 0:
res += ', "message": ['
for d in mydict:
res += '{{"label": "{}", "value": "{}"}},'.format(d,mydict[d])
res += '{{"label": "{}", "value": "{}"}},'.format(d, mydict[d])
res = res[:-1] + "]"
res += '}'
res += "}"
return res
else:
print('Invalid Input')
print("Invalid Input")
def format_str(d):
return str(d).replace("'",'"')
return str(d).replace("'", '"')

226
boards/default_src/micropython_k210_mixgoai/origin/build/lib/mpu9250.py
View File

@@ -19,22 +19,23 @@ import ustruct
import utime
import time
import math
#from machine import I2C, Pin
# from machine import I2C, Pin
# pylint: enable=import-error
__version__ = "0.2.0"
# pylint: disable=import-error
# pylint: enable=import-error
_GYRO_CONFIG = const(0x1b)
_ACCEL_CONFIG = const(0x1c)
_ACCEL_CONFIG2 = const(0x1d)
_GYRO_CONFIG = const(0x1B)
_ACCEL_CONFIG = const(0x1C)
_ACCEL_CONFIG2 = const(0x1D)
_INT_PIN_CFG = const(0x37)
_ACCEL_XOUT_H = const(0x3b)
_ACCEL_XOUT_L = const(0x3c)
_ACCEL_YOUT_H = const(0x3d)
_ACCEL_YOUT_L = const(0x3e)
_ACCEL_ZOUT_H = const(0x3f)
_ACCEL_ZOUT_L= const(0x40)
_ACCEL_XOUT_H = const(0x3B)
_ACCEL_XOUT_L = const(0x3C)
_ACCEL_YOUT_H = const(0x3D)
_ACCEL_YOUT_L = const(0x3E)
_ACCEL_ZOUT_H = const(0x3F)
_ACCEL_ZOUT_L = const(0x40)
_TEMP_OUT_H = const(0x41)
_TEMP_OUT_L = const(0x42)
_GYRO_XOUT_H = const(0x43)
@@ -45,18 +46,18 @@ _GYRO_ZOUT_H = const(0x47)
_GYRO_ZOUT_L = const(0x48)
_WHO_AM_I = const(0x75)
#_ACCEL_FS_MASK = const(0b00011000)
# _ACCEL_FS_MASK = const(0b00011000)
ACCEL_FS_SEL_2G = const(0b00000000)
ACCEL_FS_SEL_4G = const(0b00001000)
ACCEL_FS_SEL_8G = const(0b00010000)
ACCEL_FS_SEL_16G = const(0b00011000)
_ACCEL_SO_2G = 16384 # 1 / 16384 ie. 0.061 mg / digit
_ACCEL_SO_4G = 8192 # 1 / 8192 ie. 0.122 mg / digit
_ACCEL_SO_8G = 4096 # 1 / 4096 ie. 0.244 mg / digit
_ACCEL_SO_16G = 2048 # 1 / 2048 ie. 0.488 mg / digit
_ACCEL_SO_2G = 16384 # 1 / 16384 ie. 0.061 mg / digit
_ACCEL_SO_4G = 8192 # 1 / 8192 ie. 0.122 mg / digit
_ACCEL_SO_8G = 4096 # 1 / 4096 ie. 0.244 mg / digit
_ACCEL_SO_16G = 2048 # 1 / 2048 ie. 0.488 mg / digit
#_GYRO_FS_MASK = const(0b00011000)
# _GYRO_FS_MASK = const(0b00011000)
GYRO_FS_SEL_250DPS = const(0b00000000)
GYRO_FS_SEL_500DPS = const(0b00001000)
GYRO_FS_SEL_1000DPS = const(0b00010000)
@@ -73,9 +74,9 @@ _I2C_BYPASS_EN = const(0b00000010)
_I2C_BYPASS_DIS = const(0b00000000)
SF_G = 1
SF_M_S2 = 9.80665 # 1 g = 9.80665 m/s2 ie. standard gravity
SF_M_S2 = 9.80665 # 1 g = 9.80665 m/s2 ie. standard gravity
SF_DEG_S = 1
SF_RAD_S = 57.295779578552 # 1 rad/s is 57.295779578552 deg/s
SF_RAD_S = 57.295779578552 # 1 rad/s is 57.295779578552 deg/s
_WIA = const(0x00)
@@ -86,15 +87,15 @@ _HYH = const(0x06)
_HZL = const(0x07)
_HZH = const(0x08)
_ST2 = const(0x09)
_CNTL1 = const(0x0a)
_CNTL1 = const(0x0A)
_ASAX = const(0x10)
_ASAY = const(0x11)
_ASAZ = const(0x12)
_MODE_POWER_DOWN = 0b00000000
MODE_SINGLE_MEASURE = 0b00000001
MODE_CONTINOUS_MEASURE_1 = 0b00000010 # 8Hz
MODE_CONTINOUS_MEASURE_2 = 0b00000110 # 100Hz
MODE_CONTINOUS_MEASURE_1 = 0b00000010 # 8Hz
MODE_CONTINOUS_MEASURE_2 = 0b00000110 # 100Hz
MODE_EXTERNAL_TRIGGER_MEASURE = 0b00000100
_MODE_SELF_TEST = 0b00001000
_MODE_FUSE_ROM_ACCESS = 0b00001111
@@ -102,15 +103,21 @@ _MODE_FUSE_ROM_ACCESS = 0b00001111
OUTPUT_14_BIT = 0b00000000
OUTPUT_16_BIT = 0b00010000
_SO_14BIT = 0.6 # 渭T per digit when 14bit mode
_SO_16BIT = 0.15 # 渭T per digit when 16bit mode
_SO_14BIT = 0.6 # 渭T per digit when 14bit mode
_SO_16BIT = 0.15 # 渭T per digit when 16bit mode
class MPU6500:
"""Class which provides interface to MPU6500 6-axis motion tracking device."""
def __init__(
self, i2c, address=0x68,
accel_fs=ACCEL_FS_SEL_2G, gyro_fs=GYRO_FS_SEL_250DPS,
accel_sf=SF_M_S2, gyro_sf=SF_RAD_S
self,
i2c,
address=0x68,
accel_fs=ACCEL_FS_SEL_2G,
gyro_fs=GYRO_FS_SEL_250DPS,
accel_sf=SF_M_S2,
gyro_sf=SF_RAD_S,
):
self.i2c = i2c
self.address = address
@@ -125,15 +132,15 @@ class MPU6500:
# Enable I2C bypass to access for MPU9250 magnetometer access.
char = self._register_char(_INT_PIN_CFG)
char &= ~_I2C_BYPASS_MASK # clear I2C bits
char &= ~_I2C_BYPASS_MASK # clear I2C bits
char |= _I2C_BYPASS_EN
self._register_char(_INT_PIN_CFG, char)
@property
def temperature(self):
tempbuf=self._register_short(0x41)
return tempbuf/333.87 + 21 # I think
tempbuf = self._register_short(0x41)
return tempbuf / 333.87 + 21 # I think
# @property
def acceleration(self):
"""
@@ -161,7 +168,7 @@ class MPU6500:
@property
def whoami(self):
""" Value of the whoami register. """
"""Value of the whoami register."""
return self._register_char(_WHO_AM_I)
def _register_short(self, register, value=None, buf=bytearray(2)):
@@ -216,12 +223,18 @@ class MPU6500:
def __exit__(self, exception_type, exception_value, traceback):
pass
class AK8963:
"""Class which provides interface to AK8963 magnetometer."""
def __init__(
self, i2c, address=0x0c,
mode=MODE_CONTINOUS_MEASURE_1, output=OUTPUT_16_BIT,
offset=(0, 0, 0), scale=(1, 1, 1)
self,
i2c,
address=0x0C,
mode=MODE_CONTINOUS_MEASURE_1,
output=OUTPUT_16_BIT,
offset=(0, 0, 0),
scale=(1, 1, 1),
):
self.i2c = i2c
self.address = address
@@ -242,7 +255,7 @@ class AK8963:
self._adjustement = (
(0.5 * (asax - 128)) / 128 + 1,
(0.5 * (asay - 128)) / 128 + 1,
(0.5 * (asaz - 128)) / 128 + 1
(0.5 * (asaz - 128)) / 128 + 1,
)
# Power on
@@ -252,15 +265,14 @@ class AK8963:
self._so = _SO_16BIT
else:
self._so = _SO_14BIT
@property
def magnetic(self):
"""
X, Y, Z axis micro-Tesla (uT) as floats.
"""
xyz = list(self._register_three_shorts(_HXL))
self._register_char(_ST2) # Enable updating readings again
self._register_char(_ST2) # Enable updating readings again
# Apply factory axial sensitivy adjustements
xyz[0] *= self._adjustement[0]
@@ -291,7 +303,7 @@ class AK8963:
@property
def whoami(self):
""" Value of the whoami register. """
"""Value of the whoami register."""
return self._register_char(_WIA)
def calibrate(self, count=3, delay=200):
@@ -314,7 +326,6 @@ class AK8963:
maxz = max(maxz, reading[2])
count -= 1
# Hard iron correction
offset_x = (maxx + minx) / 2
offset_y = (maxy + miny) / 2
@@ -362,11 +373,12 @@ class AK8963:
def __exit__(self, exception_type, exception_value, traceback):
pass
class MPU9250:
"""Class which provides interface to MPU9250 9-axis motion tracking device."""
def __init__(self, i2c, mpu6500 = None, ak8963 = None):
def __init__(self, i2c, mpu6500=None, ak8963=None):
if mpu6500 is None:
self.mpu6500 = MPU6500(i2c)
else:
@@ -386,8 +398,8 @@ class MPU9250:
# """
# return self.mpu6500.acceleration
def mpu9250_get_temperature(self):
return self.mpu6500.temperature
return self.mpu6500.temperature
def mpu9250_get_values(self):
"""
Acceleration measured by the sensor. By default will return a
@@ -395,7 +407,7 @@ class MPU9250:
pass `accel_fs=SF_G` parameter to the MPU6500 constructor.
"""
g = self.mpu6500.acceleration()
a = [round(x/9.8, 2) for x in g]
a = [round(x / 9.8, 2) for x in g]
return tuple(a)
def mpu9250_get_x(self):
@@ -404,15 +416,15 @@ class MPU9250:
3-tuple of X, Y, Z axis values in m/s^2 as floats. To get values in g
pass `accel_fs=SF_G` parameter to the MPU6500 constructor.
"""
return round(self.mpu6500.acceleration()[0]/9.8, 2)
return round(self.mpu6500.acceleration()[0] / 9.8, 2)
def mpu9250_get_y(self):
"""
Acceleration measured by the sensor. By default will return a
3-tuple of X, Y, Z axis values in m/s^2 as floats. To get values in g
pass `accel_fs=SF_G` parameter to the MPU6500 constructor.
"""
return round(self.mpu6500.acceleration()[1]/9.8, 2)
return round(self.mpu6500.acceleration()[1] / 9.8, 2)
def mpu9250_get_z(self):
"""
@@ -420,45 +432,47 @@ class MPU9250:
3-tuple of X, Y, Z axis values in m/s^2 as floats. To get values in g
pass `accel_fs=SF_G` parameter to the MPU6500 constructor.
"""
return round(self.mpu6500.acceleration()[2]/9.8, 2)
return round(self.mpu6500.acceleration()[2] / 9.8, 2)
def mpu9250_is_gesture(self,choice):
if choice == 'face up':
if self.mpu6500.acceleration()[2] <= -9:
return True
else:
return False
if choice == 'face down':
if self.mpu6500.acceleration()[2] >= 9:
return True
else:
return False
if choice == 'shake':
if abs(self.mpu6500.acceleration()[0]) >= 9 and abs(self.mpu6500.acceleration()[1]) >= 9 :
return True
else:
return False
if choice == 'up':
if self.mpu6500.acceleration()[1] >= 9:
return True
else:
return False
if choice == 'down':
if self.mpu6500.acceleration()[1] <= -9:
return True
else:
return False
if choice == 'right':
if self.mpu6500.acceleration()[0] <= -9:
return True
else:
return False
if choice == 'left':
if self.mpu6500.acceleration()[0] >= 9:
return True
else:
return False
def mpu9250_is_gesture(self, choice):
if choice == "face up":
if self.mpu6500.acceleration()[2] <= -9:
return True
else:
return False
if choice == "face down":
if self.mpu6500.acceleration()[2] >= 9:
return True
else:
return False
if choice == "shake":
if (
abs(self.mpu6500.acceleration()[0]) >= 9
and abs(self.mpu6500.acceleration()[1]) >= 9
):
return True
else:
return False
if choice == "up":
if self.mpu6500.acceleration()[1] >= 9:
return True
else:
return False
if choice == "down":
if self.mpu6500.acceleration()[1] <= -9:
return True
else:
return False
if choice == "right":
if self.mpu6500.acceleration()[0] <= -9:
return True
else:
return False
if choice == "left":
if self.mpu6500.acceleration()[0] >= 9:
return True
else:
return False
@property
def mpu9250_gyro(self):
@@ -499,23 +513,27 @@ class MPU9250:
def mpu9250_magnetic_values(self):
return self.mpu9250_magnetic
# @property
def mpu9250_get_field_strength(self):
x=self.mpu9250_magnetic[0]
y=self.mpu9250_magnetic[1]
z=self.mpu9250_magnetic[2]
return (x**2+y**2+z**2)**0.5*1000
x = self.mpu9250_magnetic[0]
y = self.mpu9250_magnetic[1]
z = self.mpu9250_magnetic[2]
return (x**2 + y**2 + z**2) ** 0.5 * 1000
def mpu9250_heading(self):
x=self.mpu9250_magnetic[0]
y=self.mpu9250_magnetic[1]
z=self.mpu9250_magnetic[2]
a=math.atan(z/x)
b=math.atan(z/y)
xr=x*math.cos(a)+y*math.sin(a)*math.sin(b)-z*math.cos(b)*math.sin(a)
yr=x*math.cos(b)+z*math.sin(b)
return 60*math.atan(yr/xr)
x = self.mpu9250_magnetic[0]
y = self.mpu9250_magnetic[1]
z = self.mpu9250_magnetic[2]
a = math.atan(z / x)
b = math.atan(z / y)
xr = (
x * math.cos(a)
+ y * math.sin(a) * math.sin(b)
- z * math.cos(b) * math.sin(a)
)
yr = x * math.cos(b) + z * math.sin(b)
return 60 * math.atan(yr / xr)
@property
def whoami(self):
@@ -527,6 +545,7 @@ class MPU9250:
def __exit__(self, exception_type, exception_value, traceback):
pass
class Compass:
RAD_TO_DEG = 57.295779513082320876798154814105
@@ -547,20 +566,23 @@ class Compass:
def heading(self):
from math import atan2
xyz = self.sensor.mpu9250_magnetic
return int(((atan2(xyz[1], xyz[0]) * Compass.RAD_TO_DEG) + 180) % 360)
def is_calibrate(self):
try:
import compass_cfg
return True
except Exception as e:
return False
def reset_calibrate(self):
import os
os.remove("compass_cfg.py")
# compass = mpu
# accelerometer = mpu

241
boards/default_src/micropython_k210_mixgoai/origin/build/lib/ms32006.py
View File

@@ -8,135 +8,136 @@ Micropython library for the MS32006 step diever
dahanzimin From the Mixly Team
"""
import time
from micropython import const
MS32006_REG_RESET = const(0x00) #复位
MS32006_FCLK = const(25000000) #芯片输入时钟选择，此参数与运动速度有关。 范围是：5-30MHZ
MS32006_REG_RESET = const(0x00) # 复位
MS32006_FCLK = const(
25000000
) # 芯片输入时钟选择，此参数与运动速度有关。 范围是：5-30MHZ
ADDRESS_A = 0x10
ADDRESS_B = 0x18
MOT_FULL = 0
MOT_HALF = 1
MOT_A = 0
MOT_B = 4
MOT_N = 0
MOT_CW = 1
MOT_CCW = 2
MOT_P = 3
ADDRESS_A = 0x10
ADDRESS_B = 0x18
MOT_FULL = 0
MOT_HALF = 1
MOT_A = 0
MOT_B = 4
MOT_N = 0
MOT_CW = 1
MOT_CCW = 2
MOT_P = 3
class MS32006:
_buffer = bytearray(2)
_buffer = bytearray(2)
def __init__(self, i2c_bus,addr=ADDRESS_A,mode=MOT_FULL):
self._device = i2c_bus
self._address = addr
self.reset()
self.mode=mode
def __init__(self, i2c_bus, addr=ADDRESS_A, mode=MOT_FULL):
self._device = i2c_bus
self._address = addr
self.reset()
self.mode = mode
def _read_u8(self, address):
self._buffer[0] = address & 0xFF
self._device.writeto(self._address,self._buffer)
self._device.readfrom_into(self._address,self._buffer)
return self._buffer[0]
def _write_u8(self, address, val):
self._buffer[0] = address & 0xFF
self._buffer[1] = val & 0xFF
self._device.writeto(self._address,self._buffer)
def _read_u8(self, address):
self._buffer[0] = address & 0xFF
self._device.writeto(self._address, self._buffer)
self._device.readfrom_into(self._address, self._buffer)
return self._buffer[0]
def reset(self):
self._write_u8(MS32006_REG_RESET,0x00)
time.sleep(0.1)
self._write_u8(MS32006_REG_RESET,0xC1)
def move(self,moto,mot_dir,mot_pps,mot_step):
readstate_0H = self._read_u8(0x00)
readstate_9H = self._read_u8(0x09)
speed_data=MS32006_FCLK//mot_pps//128 #设置速度 xx pps 128是固定参数
if speed_data<32: #限定转速
speed_data=32
elif speed_data>16383:
speed_data=16383
mot_speed_l=speed_data&0x00ff #取低8位
mot_speed_h=speed_data//0x100 #取高6位
if self.mode==MOT_FULL: #设置整步、半步驱动模式
mot_speed_h|=0x80
else:
mot_speed_h&=0x7f
if mot_step>2047:
raise AttributeError("Reach the set upper limit, up to 2047 step")
mot_step_l=mot_step&0x00ff
mot_step_h=mot_step//0x100
mot_step_h|=0x80
if mot_dir==MOT_CW:
mot_step_h&=0xBF
else:
mot_step_h|=0x40
self._write_u8(0x01+moto,mot_speed_l)
self._write_u8(0x02+moto,mot_speed_h)
self._write_u8(0x03+moto,mot_step_l)
self._write_u8(0x04+moto,mot_step_h)
if moto==MOT_A:
self._write_u8(0x00, readstate_0H&0xfb)
self._write_u8(0x09, readstate_9H|0x80)
else:
self._write_u8(0x00, readstate_0H&0xfd)
self._write_u8(0x09, readstate_9H|0x40)
def close(self,moto): #停止并关闭输出
if moto==MOT_A:
self._write_u8(0x04,0x00)
else:
self._write_u8(0x08,0x00)
def _write_u8(self, address, val):
self._buffer[0] = address & 0xFF
self._buffer[1] = val & 0xFF
self._device.writeto(self._address, self._buffer)
def stop(self,moto): #此停止函数，强制让电机停止
readstate = self._read_u8(0x00)
if moto==MOT_A:
self._write_u8(0x00,readstate|0x04)
else:
self._write_u8(0x00,readstate|0x02)
def readstep(self,moto): #读取电机运动步数
if moto==MOT_A:
rdb =self._read_u8(0x0b)
rdc =self._read_u8(0x0c)
else:
rdb =self._read_u8(0x0d)
rdc =self._read_u8(0x0e)
return (rdb*0x100+rdc)&0xfff
def reset(self):
self._write_u8(MS32006_REG_RESET, 0x00)
time.sleep(0.1)
self._write_u8(MS32006_REG_RESET, 0xC1)
def readbusy(self,moto): #读取电机缓存是否有数据
if moto==MOT_A:
busy =(self._read_u8(0x0b)>>6)&1
else:
busy =(self._read_u8(0x0d)>>6)&1
return bool(busy)
def readwork(self,moto): #读取电机是否在运行
if moto==MOT_A:
busy =(self._read_u8(0x0b)>>4)&1
else:
busy =(self._read_u8(0x0d)>>4)&1
return bool(busy)
def dc_motor(self,state,speed): #直流电机驱动
if (state==MOT_CW) | (state==MOT_CCW) :
speed_st=speed*127//100 |0x80
self._write_u8(0x0A,speed_st)
readstate = self._read_u8(0x09) & 0xA0
state_st=(state<<2) | 0X03 | readstate
self._write_u8(0x09,state_st)
def move(self, moto, mot_dir, mot_pps, mot_step):
readstate_0H = self._read_u8(0x00)
readstate_9H = self._read_u8(0x09)
speed_data = MS32006_FCLK // mot_pps // 128 # 设置速度 xx pps 128是固定参数
if speed_data < 32: # 限定转速
speed_data = 32
elif speed_data > 16383:
speed_data = 16383
mot_speed_l = speed_data & 0x00FF # 取低8位
mot_speed_h = speed_data // 0x100 # 取高6位
if self.mode == MOT_FULL: # 设置整步、半步驱动模式
mot_speed_h |= 0x80
else:
mot_speed_h &= 0x7F
if mot_step > 2047:
raise AttributeError("Reach the set upper limit, up to 2047 step")
mot_step_l = mot_step & 0x00FF
mot_step_h = mot_step // 0x100
mot_step_h |= 0x80
if mot_dir == MOT_CW:
mot_step_h &= 0xBF
else:
mot_step_h |= 0x40
self._write_u8(0x01 + moto, mot_speed_l)
self._write_u8(0x02 + moto, mot_speed_h)
self._write_u8(0x03 + moto, mot_step_l)
self._write_u8(0x04 + moto, mot_step_h)
if moto == MOT_A:
self._write_u8(0x00, readstate_0H & 0xFB)
self._write_u8(0x09, readstate_9H | 0x80)
else:
self._write_u8(0x00, readstate_0H & 0xFD)
self._write_u8(0x09, readstate_9H | 0x40)
def close(self, moto): # 停止并关闭输出
if moto == MOT_A:
self._write_u8(0x04, 0x00)
else:
self._write_u8(0x08, 0x00)
def stop(self, moto): # 此停止函数，强制让电机停止
readstate = self._read_u8(0x00)
if moto == MOT_A:
self._write_u8(0x00, readstate | 0x04)
else:
self._write_u8(0x00, readstate | 0x02)
def readstep(self, moto): # 读取电机运动步数
if moto == MOT_A:
rdb = self._read_u8(0x0B)
rdc = self._read_u8(0x0C)
else:
rdb = self._read_u8(0x0D)
rdc = self._read_u8(0x0E)
return (rdb * 0x100 + rdc) & 0xFFF
def readbusy(self, moto): # 读取电机缓存是否有数据
if moto == MOT_A:
busy = (self._read_u8(0x0B) >> 6) & 1
else:
busy = (self._read_u8(0x0D) >> 6) & 1
return bool(busy)
def readwork(self, moto): # 读取电机是否在运行
if moto == MOT_A:
busy = (self._read_u8(0x0B) >> 4) & 1
else:
busy = (self._read_u8(0x0D) >> 4) & 1
return bool(busy)
def dc_motor(self, state, speed): # 直流电机驱动
if (state == MOT_CW) | (state == MOT_CCW):
speed_st = speed * 127 // 100 | 0x80
self._write_u8(0x0A, speed_st)
readstate = self._read_u8(0x09) & 0xA0
state_st = (state << 2) | 0x03 | readstate
self._write_u8(0x09, state_st)

59
boards/default_src/micropython_k210_mixgoai/origin/build/lib/net_espat.py
View File

@@ -1,30 +1,31 @@
import network,time,board
import network, time, board
from machine import UART
import time
def wifi_init(RX_Pin,TX_Pin):
board.register(TX_Pin,board.FPIOA.UART2_TX)
board.register(RX_Pin,board.FPIOA.UART2_RX)
uart = UART(UART.UART2,115200,timeout=1000, read_buf_len=1024*16)
T1=time.ticks_ms()
while True:
tmp=uart.read()
if tmp:
if tmp.endswith("OK\r\n"):
break
else:
uart.write("AT+RST\r\n")
time.sleep_ms(20)
if time.ticks_diff(time.ticks_ms(), T1) >2000:
raise AttributeError("ESP-AT not connected or needs to be reset")
try:
nic = network.ESP8285(uart)
time.sleep(1)
print("ESP-AT OK")
return nic
except Exception:
raise AttributeError("ESP-AT Connection Failed")
def wifi_init(RX_Pin, TX_Pin):
board.register(TX_Pin, board.FPIOA.UART2_TX)
board.register(RX_Pin, board.FPIOA.UART2_RX)
uart = UART(UART.UART2, 115200, timeout=1000, read_buf_len=1024 * 16)
T1 = time.ticks_ms()
while True:
tmp = uart.read()
if tmp:
if tmp.endswith("OK\r\n"):
break
else:
uart.write("AT+RST\r\n")
time.sleep_ms(20)
if time.ticks_diff(time.ticks_ms(), T1) > 2000:
raise AttributeError("ESP-AT not connected or needs to be reset")
try:
nic = network.ESP8285(uart)
time.sleep(1)
print("ESP-AT OK")
return nic
except Exception:
raise AttributeError("ESP-AT Connection Failed")
def wifi_deal_ap_info(info):
@@ -40,9 +41,9 @@ def wifi_deal_ap_info(info):
res.append(info_one)
return res
def scans(nic):
ap_info = nic.scan()
ap_info = wifi_deal_ap_info(ap_info)
ap_info.sort(key=lambda x:x[2], reverse=True)
return ap_info
def scans(nic):
ap_info = nic.scan()
ap_info = wifi_deal_ap_info(ap_info)
ap_info.sort(key=lambda x: x[2], reverse=True)
return ap_info

22
boards/default_src/micropython_k210_mixgoai/origin/build/lib/pid.py
View File

@@ -1,16 +1,18 @@
import time
from math import pi, isnan
class PID:
_kp = _ki = _kd = _integrator = _imax = 0
_last_error = _last_derivative = _last_t = 0
_RC = 1/(2 * pi * 20)
_RC = 1 / (2 * pi * 20)
def __init__(self, p=0, i=0, d=0, imax=0):
self._kp = float(p)
self._ki = float(i)
self._kd = float(d)
self._imax = abs(imax)
self._last_derivative = float('nan')
self._last_derivative = float("nan")
def get_pid(self, error, scaler):
tnow = time.ticks_ms()
@@ -28,19 +30,23 @@ class PID:
self._last_derivative = 0
else:
derivative = (error - self._last_error) / delta_time
derivative = self._last_derivative + \
((delta_time / (self._RC + delta_time)) * \
(derivative - self._last_derivative))
derivative = self._last_derivative + (
(delta_time / (self._RC + delta_time))
* (derivative - self._last_derivative)
)
self._last_error = error
self._last_derivative = derivative
output += self._kd * derivative
output *= scaler
if abs(self._ki) > 0 and dt > 0:
self._integrator += (error * self._ki) * scaler * delta_time
if self._integrator < -self._imax: self._integrator = -self._imax
elif self._integrator > self._imax: self._integrator = self._imax
if self._integrator < -self._imax:
self._integrator = -self._imax
elif self._integrator > self._imax:
self._integrator = self._imax
output += self._integrator
return output
def reset_I(self):
self._integrator = 0
self._last_derivative = float('nan')
self._last_derivative = float("nan")

261
boards/default_src/micropython_k210_mixgoai/origin/build/lib/player.py
View File

@@ -1,128 +1,151 @@
import board
import audio,video
import audio, video
from Maix import I2S
import gc
spk_b=None
spk_d=None
spk_w=None
def spk_init(BLK=8,WS=9,DAT=10,sample_rate=16000):
global spk_b
global spk_d
global spk_w
spk_b=BLK
spk_d=DAT
spk_w=WS
board.register(DAT,board.FPIOA.I2S0_OUT_D1)
board.register(BLK,board.FPIOA.I2S0_SCLK)
board.register(WS,board.FPIOA.I2S0_WS)
wav_dev = I2S(I2S.DEVICE_0)
wav_dev.channel_config(I2S.CHANNEL_1, I2S.TRANSMITTER,resolution = I2S.RESOLUTION_16_BIT, cycles = I2S.SCLK_CYCLES_32, align_mode = I2S.STANDARD_MODE)
wav_dev.set_sample_rate(sample_rate)
spk_rep=wav_dev
return wav_dev
def mic_init(BLK=35,WS=33,DAT=34,sample_rate=16000):
board.register(DAT,board.FPIOA.I2S2_IN_D0)
board.register(BLK,board.FPIOA.I2S2_SCLK)
board.register(WS,board.FPIOA.I2S2_WS)
wav_dev = I2S(I2S.DEVICE_2)
wav_dev.channel_config(I2S.CHANNEL_0, I2S.RECEIVER, resolution = I2S.RESOLUTION_16_BIT, cycles = I2S.SCLK_CYCLES_32, align_mode=I2S.STANDARD_MODE)
wav_dev.set_sample_rate(sample_rate)
return wav_dev
spk_b = None
spk_d = None
spk_w = None
def audio_play(I2S,path,num=80):
try:
player = audio.Audio(path=path)
except Exception as e:
raise NameError("No audio file loaded or {}".format(e))
player.volume(num)
wav_info=player.play_process(I2S)
I2S.set_sample_rate(wav_info[1])
while True:
ret = player.play()
if ret == None:
print("Format Error")
break
elif ret == 0:
print("Play end \n")
player.finish()
break
player.__deinit__()
gc.collect()
def audio_record(I2S,path,record_time,sample_rate=16000):
try:
recorder = audio.Audio(path=path, is_create=True, samplerate=sample_rate)
except Exception as e:
raise NameError("Need audio storage location or {}".format(e))
queue = []
frame_cnt = record_time*sample_rate//2048
for i in range(frame_cnt):
tmp = I2S.record(2048*2)
if len(queue) > 0:
ret = recorder.record(queue[0])
queue.pop(0)
I2S.wait_record()
queue.append(tmp)
print("record:{}s".format(round(((frame_cnt-i-1)/7.7) ,1)))
recorder.finish()
recorder.__deinit__()
del recorder
print("Audio record finish \n")
gc.collect()
def video_play(I2S1,path,num=80):
try:
global spk_b
global spk_d
global spk_w
import lcd
lcd.init()
I2S=spk_init(spk_b,spk_w,spk_d)
vide = video.open(path)
except Exception as e:
raise NameError("No video file loaded or {}".format(e))
vide.volume(num)
while True:
try:
ret = vide.play()
except Exception as e:
raise NameError("Video format error or {}".format(e))
if ret == None:
print("Format Error")
break
elif ret == 0:
print("Play end \n")
break
vide.__del__()
del vide
del I2S
gc.collect()
def spk_init(BLK=8, WS=9, DAT=10, sample_rate=16000):
global spk_b
global spk_d
global spk_w
spk_b = BLK
spk_d = DAT
spk_w = WS
board.register(DAT, board.FPIOA.I2S0_OUT_D1)
board.register(BLK, board.FPIOA.I2S0_SCLK)
board.register(WS, board.FPIOA.I2S0_WS)
wav_dev = I2S(I2S.DEVICE_0)
wav_dev.channel_config(
I2S.CHANNEL_1,
I2S.TRANSMITTER,
resolution=I2S.RESOLUTION_16_BIT,
cycles=I2S.SCLK_CYCLES_32,
align_mode=I2S.STANDARD_MODE,
)
wav_dev.set_sample_rate(sample_rate)
spk_rep = wav_dev
return wav_dev
def video_record(I2S,path,record_time):
import sensor,lcd
lcd.init()
try:
v = video.open(path, audio=False, record=True, interval=200000, quality=80,width=240, height=240)
except Exception as e:
raise NameError("Need video storage location or {}".format(e))
record_time=record_time*5
for i in range(record_time):
try:
img = sensor.snapshot()
except :
raise NameError("Need to initialize camera")
lcd.display(img)
v.record(img)
print("record {}s".format(round((record_time-i-1)*0.2,1)))
v.record_finish()
print("Video record finish \n")
v.__del__()
gc.collect()
def mic_init(BLK=35, WS=33, DAT=34, sample_rate=16000):
board.register(DAT, board.FPIOA.I2S2_IN_D0)
board.register(BLK, board.FPIOA.I2S2_SCLK)
board.register(WS, board.FPIOA.I2S2_WS)
wav_dev = I2S(I2S.DEVICE_2)
wav_dev.channel_config(
I2S.CHANNEL_0,
I2S.RECEIVER,
resolution=I2S.RESOLUTION_16_BIT,
cycles=I2S.SCLK_CYCLES_32,
align_mode=I2S.STANDARD_MODE,
)
wav_dev.set_sample_rate(sample_rate)
return wav_dev
def audio_play(I2S, path, num=80):
try:
player = audio.Audio(path=path)
except Exception as e:
raise NameError("No audio file loaded or {}".format(e))
player.volume(num)
wav_info = player.play_process(I2S)
I2S.set_sample_rate(wav_info[1])
while True:
ret = player.play()
if ret == None:
print("Format Error")
break
elif ret == 0:
print("Play end \n")
player.finish()
break
player.__deinit__()
gc.collect()
def audio_record(I2S, path, record_time, sample_rate=16000):
try:
recorder = audio.Audio(path=path, is_create=True, samplerate=sample_rate)
except Exception as e:
raise NameError("Need audio storage location or {}".format(e))
queue = []
frame_cnt = record_time * sample_rate // 2048
for i in range(frame_cnt):
tmp = I2S.record(2048 * 2)
if len(queue) > 0:
ret = recorder.record(queue[0])
queue.pop(0)
I2S.wait_record()
queue.append(tmp)
print("record:{}s".format(round(((frame_cnt - i - 1) / 7.7), 1)))
recorder.finish()
recorder.__deinit__()
del recorder
print("Audio record finish \n")
gc.collect()
def video_play(I2S1, path, num=80):
try:
global spk_b
global spk_d
global spk_w
import lcd
lcd.init()
I2S = spk_init(spk_b, spk_w, spk_d)
vide = video.open(path)
except Exception as e:
raise NameError("No video file loaded or {}".format(e))
vide.volume(num)
while True:
try:
ret = vide.play()
except Exception as e:
raise NameError("Video format error or {}".format(e))
if ret == None:
print("Format Error")
break
elif ret == 0:
print("Play end \n")
break
vide.__del__()
del vide
del I2S
gc.collect()
def video_record(I2S, path, record_time):
import sensor, lcd
lcd.init()
try:
v = video.open(
path,
audio=False,
record=True,
interval=200000,
quality=80,
width=240,
height=240,
)
except Exception as e:
raise NameError("Need video storage location or {}".format(e))
record_time = record_time * 5
for i in range(record_time):
try:
img = sensor.snapshot()
except:
raise NameError("Need to initialize camera")
lcd.display(img)
v.record(img)
print("record {}s".format(round((record_time - i - 1) * 0.2, 1)))
v.record_finish()
print("Video record finish \n")
v.__del__()
gc.collect()

56
boards/default_src/micropython_k210_mixgoai/origin/build/lib/reset.py
View File

@@ -1,33 +1,35 @@
try:
import image
image.font_free()
import image
image.font_free()
except:
pass
try:
import lcd,time,gc,machine
lcd.init(color=0x0000)
lcd.draw_string(48,100, "Welcome to MixGo!", lcd.YELLOW, lcd.BLACK)
lcd.draw_string(62,132, "loading .", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading ..", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading ...", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading ....", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading .....", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading ......", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62,132, "loading .......", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.clear(0x0000)
del time
del lcd
del gc
import lcd, time, gc, machine
lcd.init(color=0x0000)
lcd.draw_string(48, 100, "Welcome to MixGo!", lcd.YELLOW, lcd.BLACK)
lcd.draw_string(62, 132, "loading .", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading ..", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading ...", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading ....", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading .....", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading ......", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.draw_string(62, 132, "loading .......", lcd.YELLOW, lcd.BLACK)
time.sleep_ms(50)
lcd.clear(0x0000)
del time
del lcd
del gc
finally:
import gc
gc.collect()
import gc
gc.collect()

28
boards/default_src/micropython_k210_mixgoai/origin/build/lib/servo.py
View File

@@ -2,6 +2,8 @@
import math
import ustruct
import time
class PCA9685:
def __init__(self, i2c, address=0x40):
self.i2c = i2c
@@ -15,25 +17,25 @@ class PCA9685:
return self.i2c.readfrom_mem(self.address, address, 1)[0]
def reset(self):
self._write(0x00, 0x00) # Mode1
self._write(0x00, 0x00) # Mode1
def freq(self, freq=None):
if freq is None:
return int(25000000.0 / 4096 / (self._read(0xfe) - 0.5))
return int(25000000.0 / 4096 / (self._read(0xFE) - 0.5))
prescale = int(25000000.0 / 4096.0 / freq + 0.5)
old_mode = self._read(0x00) # Mode 1
self._write(0x00, (old_mode & 0x7F) | 0x10) # Mode 1, sleep
self._write(0xfe, prescale) # Prescale
self._write(0x00, old_mode) # Mode 1
old_mode = self._read(0x00) # Mode 1
self._write(0x00, (old_mode & 0x7F) | 0x10) # Mode 1, sleep
self._write(0xFE, prescale) # Prescale
self._write(0x00, old_mode) # Mode 1
time.sleep_us(5)
self._write(0x00, old_mode | 0xa1) # Mode 1, autoincrement on
self._write(0x00, old_mode | 0xA1) # Mode 1, autoincrement on
def pwm(self, index, on=None, off=None):
if on is None or off is None:
data = self.i2c.readfrom_mem(self.address, 0x06 + 4 * index, 4)
return ustruct.unpack('<HH', data)
data = ustruct.pack('<HH', on, off)
self.i2c.writeto_mem(self.address, 0x06 + 4 * index, data)
return ustruct.unpack("<HH", data)
data = ustruct.pack("<HH", on, off)
self.i2c.writeto_mem(self.address, 0x06 + 4 * index, data)
def duty(self, index, value=None, invert=False):
if value is None:
@@ -57,9 +59,11 @@ class PCA9685:
else:
self.pwm(index, 0, value)
class Servos:
def __init__(self, i2c, address=0x55, freq=50, min_us=600, max_us=2400,
degrees=180):
def __init__(
self, i2c, address=0x55, freq=50, min_us=600, max_us=2400, degrees=180
):
self.period = 1000000 / freq
self.min_duty = self._us2duty(min_us)
self.max_duty = self._us2duty(max_us)

18
boards/default_src/micropython_k210_mixgoai/origin/build/lib/sht20.py
View File

@@ -5,12 +5,11 @@ from time import sleep_ms
SHT20_I2CADDR = 64
# SHT20 Command
TRI_T_MEASURE_NO_HOLD = b'\xf3'
TRI_RH_MEASURE_NO_HOLD = b'\xf5'
READ_USER_REG = b'\xe7'
WRITE_USER_REG = b'\xe6'
SOFT_RESET = b'\xfe'
TRI_T_MEASURE_NO_HOLD = b"\xf3"
TRI_RH_MEASURE_NO_HOLD = b"\xf5"
READ_USER_REG = b"\xe7"
WRITE_USER_REG = b"\xe6"
SOFT_RESET = b"\xfe"
class SHT20(object):
@@ -23,7 +22,7 @@ class SHT20(object):
self._bus.writeto(self._address, TRI_T_MEASURE_NO_HOLD)
sleep_ms(150)
origin_data = self._bus.readfrom(self._address, 2)
origin_value = unp('>h', origin_data)[0]
origin_value = unp(">h", origin_data)[0]
value = -46.85 + 175.72 * (origin_value / 65536)
return value
@@ -31,8 +30,9 @@ class SHT20(object):
self._bus.writeto(self._address, TRI_RH_MEASURE_NO_HOLD)
sleep_ms(150)
origin_data = self._bus.readfrom(self._address, 2)
origin_value = unp('>H', origin_data)[0]
origin_value = unp(">H", origin_data)[0]
value = -6 + 125 * (origin_value / 65536)
return value
#sht=SHT20(I2C(scl = Pin(22), sda = Pin(21), freq = 100000))
# sht=SHT20(I2C(scl = Pin(22), sda = Pin(21), freq = 100000))