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

全部mpy增加心知天气、网络请求、ds18x20

Browse Source
This commit is contained in:
Irene-Maxine
2024-09-28 00:08:39 +08:00
parent 1d2957442b
commit 93cb1f887e
66 changed files with 2992 additions and 1194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
boards/default/micropython/build/lib/ds18b20.py Normal file
View File

@@ -0,0 +1,65 @@
# DS18x20 temperature sensor driver for MicroPython.
# MIT license; Copyright (c) 2016 Damien P. George
import onewire
from machine import Pin
from micropython import const
_CONVERT = const(0x44)
_RD_SCRATCH = const(0xBE)
_WR_SCRATCH = const(0x4E)
class DS18X20:
__species = {}
__first_init = True
def __new__(cls, pin, *args, **kwargs):
if pin not in cls.__species.keys():
cls.__first_init = True
cls.__species[pin] = object.__new__(cls)
return cls.__species[pin]
def __init__(self, pin):
if self.__first_init:
self.__first_init = False
self._ow = onewire.OneWire(Pin(pin, pull=Pin.PULL_UP))
self._buf = bytearray(9)
self._roms = self.scan()
def scan(self):
return [rom for rom in self._ow.scan() if rom[0] in (0x10, 0x22, 0x28)]
def convert_temp(self):
self._ow.reset(True)
self._ow.writebyte(self._ow.SKIP_ROM)
self._ow.writebyte(_CONVERT)
def read_scratch(self, rom):
self._ow.reset(True)
self._ow.select_rom(rom)
self._ow.writebyte(_RD_SCRATCH)
self._ow.readinto(self._buf)
if self._ow.crc8(self._buf):
raise Exception("CRC error")
return self._buf
def read_temp(self, rom):
buf = self.read_scratch(rom)
if rom[0] == 0x10:
if buf[1]:
t = buf[0] >> 1 | 0x80
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)
return t / 16
def temperature(self):
temp = []
self.convert_temp()
for rom in self._roms:
temp.append(round(self.read_temp(rom), 2))
return temp [0] if len(temp) == 1 else tuple(temp)

269
boards/default/micropython/build/lib/seniverse_api.py
View File

@@ -20,115 +20,198 @@ _weather_alarm="http://api.seniverse.com/v3/weather/alarm.json?" #气象灾害
_life_suggestion="http://api.seniverse.com/v3/life/suggestion.json?" #生活指数
_air_now="http://api.seniverse.com/v3/air/now.json?" #空气质量实况
_air_daily="http://api.seniverse.com/v3/air/daily.json?" #逐日空气质量预报
_tide_daily="http://api.seniverse.com/v3/tide/daily.json?" #逐小时潮汐预报
_air_hourly="http://api.seniverse.com/v3/air/hourly.json?" #逐时空气质量预报
_tide_daily="http://api.seniverse.com/v3/tide/daily.json?" #逐时潮汐预报
_geo_sun="http://api.seniverse.com/v3/geo/sun.json?" #日出日落
_geo_moon="http://api.seniverse.com/v3/geo/moon.json?" #月出月落和月相
_location_search="http://api.seniverse.com/v3/location/search.json?" #城市搜索
#数据请求
def _urequests_api(url):
try:
results=json.loads(urequests.post(url).text)
except Exception as e:
raise RuntimeError("API request failed or WiFi is not connected",e)
if "status" in results.keys():
raise ValueError(results["status"])
if "results" in results.keys():
return results["results"]
#天气实况 https://docs.seniverse.com/api/weather/now.html
def weather_now(key,location):
url="{}key={}&location={}".format(_weather_now,key,location)
results=_urequests_api(url)[0]
return results['now']
class API_BASE:
_results = None
def _urequests_api(self, url):
try:
results=json.loads(urequests.post(url).text)
except Exception as e:
raise RuntimeError("API request failed or WiFi is not connected",e)
if "status" in results.keys():
raise ValueError(results["status"])
if "results" in results.keys():
return results["results"]
#天气实况 https://docs.seniverse.com/api/weather/now.html
class Weather_now(API_BASE):
def request(self, key, location):
url = "{}key={}&location={}".format(_weather_now, key, location)
self._results = self._urequests_api(url)[0]['now']
def analysis(self, key=None):
if key is None:
return self._results
else:
return self._results[key]
Weather_now = Weather_now()
#逐日天气预报 https://docs.seniverse.com/api/weather/daily.html
def weather_daily(key,location,days=1):
url="{}key={}&location={}&days={}".format(_weather_daily,key,location,days)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['daily'])):
data.append(results['daily'][i])
return tuple(data)
class Weather_daily(API_BASE):
def request(self, key, location, days=1):
url = "{}key={}&location={}&days={}".format(_weather_daily, key, location, days)
self._results = self._urequests_api(url)[0]['daily']
def analysis(self, days=1, key=None):
if key is None:
return self._results[days]
else:
return self._results[days][key]
Weather_daily = Weather_daily()
#逐时天气预报 https://docs.seniverse.com/api/weather/hourly.html
def weather_hourly(key,location,hours=1):
url="{}key={}&location={}&hours={}".format(_weather_hourly,key,location,hours)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['hourly'])):
data.append(results['hourly'][i])
return tuple(data)
class Weather_hourly(API_BASE):
def request(self, key, location, hours=1):
url = "{}key={}&location={}&hours={}".format(_weather_hourly, key, location, hours)
self._results = self._urequests_api(url)[0]['hourly']
#气象灾害预警 https://docs.seniverse.com/api/weather/alarm.html
def weather_alarm(key,location):
url="{}key={}&location={}".format(_weather_alarm,key,location)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['alarms'])):
data.append(results['alarms'][i])
return tuple(data)
def analysis(self, hours=1, key=None):
if key is None:
return self._results[hours]
else:
return self._results[hours][key]
#生活指数 https://docs.seniverse.com/api/life/suggestion.html
def life_suggestion(key,location,days=1):
url="{}key={}&location={}&days={}".format(_life_suggestion,key,location,days)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['suggestion'])):
data.append(results['suggestion'][i])
return tuple(data)
#Weather_hourly = Weather_hourly() #暂不开启
#空气质量实况 https://docs.seniverse.com/api/air/now.html
def air_now(key,location):
url="{}key={}&location={}&scope=city".format(_air_now,key,location)
results=_urequests_api(url)[0]
return results['air']['city']
#逐日空气质量预报 https://docs.seniverse.com/api/air/daily5d.html
def air_daily(key,location,days=1):
url="{}key={}&location={}&days={}".format(_air_daily,key,location,days)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['daily'])):
data.append(results['daily'][i])
return tuple(data)
class Air_now(API_BASE):
def request(self, key, location):
url = "{}key={}&location={}&scope=city".format(_air_now, key, location)
self._results = self._urequests_api(url)[0]['air']['city']
#逐时潮汐预报 https://docs.seniverse.com/api/ocean/tide.html
def tide_daily(key,location):
url="{}key={}&location={}&days=1".format(_tide_daily,key,location)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['ports'])):
data.append({'port':results['ports'][i]['port'],
'tide':results['ports'][i]['data'][0]['tide'],
'range':results['ports'][i]['data'][0]['range']})
return tuple(data)
def analysis(self, key=None):
if key is None:
return self._results
else:
return self._results[key]
Air_now = Air_now()
#逐日空气质量预报 https://docs.seniverse.com/api/air/daily5d.html
class Air_daily(API_BASE):
def request(self, key, location, days=1):
url = "{}key={}&location={}&days={}".format(_air_daily, key, location, days)
self._results = self._urequests_api(url)[0]['daily']
def analysis(self, days=1, key=None):
if key is None:
return self._results[days]
else:
return self._results[days][key]
Air_daily= Air_daily()
#逐时空气质量预报 https://docs.seniverse.com/api/air/hourly5d.html
class Air_hourly(API_BASE):
def request(self, key, location, hours=1):
url = "{}key={}&location={}&hours={}&days=1".format(_air_hourly, key, location, hours)
self._results = self._urequests_api(url)[0]['hourly']
def analysis(self, hours=1, key=None):
if key is None:
return self._results[hours]
else:
return self._results[hours][key]
#Air_hourly = Air_hourly() #暂不开启
#气象灾害预警 https://docs.seniverse.com/api/weather/alarm.html
class Weather_alarm(API_BASE):
def request(self, key, location):
url = "{}key={}&location={}".format(_weather_alarm, key, location)
results = self._urequests_api(url)[0]['alarms']
self._results = results[0] if results else {}
def analysis(self, key=None):
if key is None:
return self._results
if key in self._results.keys():
return self._results[key]
Weather_alarm = Weather_alarm()
#生活指数 https://docs.seniverse.com/api/life/suggestion.html
class Life_suggestion(API_BASE):
def request(self, key, location):
url = "{}key={}&location={}".format(_life_suggestion, key, location)
self._results = self._urequests_api(url)[0]['suggestion']
def analysis(self, key=None, brief=False):
if key is None:
return self._results
else:
return self._results[key]['brief'] if brief else self._results[key]['details']
Life_suggestion = Life_suggestion()
#24时潮汐预报 https://docs.seniverse.com/api/ocean/tide.html
class Tide_daily(API_BASE):
def request(self, key, location):
url = "{}key={}&location={}&days=1".format(_tide_daily, key, location)
self._results = self._urequests_api(url)[0]['ports'][0]['data'][0]
def analysis(self, key=None):
if key is None:
return self._results
else:
key = key.split(',')
if len(key) == 1:
return self._results[key[0]]
if len(key) == 2:
return self._results['range'][int(key[0])][key[1]]
Tide_daily = Tide_daily()
#日出日落 https://docs.seniverse.com/api/geo/sun.html
def geo_sun(key,location,days=1):
url="{}key={}&location={}&days={}".format(_geo_sun,key,location,days)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['sun'])):
data.append(results['sun'][i])
return tuple(data)
class Geo_sun(API_BASE):
def request(self, key, location, days=1):
url = "{}key={}&location={}&days={}".format(_geo_sun, key, location, days)
self._results = self._urequests_api(url)[0]['sun']
def analysis(self, days=1, key=None):
if key is None:
return self._results[days]
else:
return self._results[days][key]
Geo_sun = Geo_sun()
#月出月落和月相 https://docs.seniverse.com/api/geo/moon.html
def geo_moon(key,location,days=1):
url="{}key={}&location={}&days={}".format(_geo_moon,key,location,days)
results=_urequests_api(url)[0]
data=[]
for i in range(len(results['moon'])):
data.append(results['moon'][i])
return tuple(data)
class Geo_moon(API_BASE):
def request(self, key, location, days=1):
url = "{}key={}&location={}&days={}".format(_geo_moon, key, location, days)
self._results = self._urequests_api(url)[0]['moon']
def analysis(self, days=1, key=None):
if key is None:
return self._results[days]
else:
return self._results[days][key]
Geo_moon = Geo_moon()
#城市搜索 https://docs.seniverse.com/api/fct/search.html
def location_search(key,location):
url="{}key={}&q={}&limit=50".format(_location_search,key,location)
results=_urequests_api(url)
data=[]
for i in range(len(results)):
data.append(results[i])
return tuple(data)
class Location_search(API_BASE):
def request(self, key, location):
url = "{}key={}&q={}&limit=50".format(_location_search, key, location)
results = self._urequests_api(url)
self._results = results[0] if results else {}
def analysis(self, key=None):
if key is None:
return self._results
else:
if key in self._results.keys():
return self._results[key]
Location_search = Location_search()

230
boards/default/micropython/build/lib/spl06_001.py
View File

@@ -1,115 +1,115 @@
"""
_SPL06-001
MicroPython library for the _SPL06-001(Air pressure sensor)
=======================================================
@dahanzimin From the Mixly Team
"""
import time
from micropython import const
_SPL06_ADDRESS = const(0x77)
_SPL06_REG_PSR = const(0x00)
_SPL06_REG_TMP = const(0x03)
_SPL06_PSR_CFG = const(0x06)
_SPL06_TMP_CFG = const(0x07)
_SPL06_MEAS_CFG = const(0x08)
_SPL06_CFG_REG = const(0x09)
_SPL06_REG_RST = const(0x0C)
_SPL06_REG_ID = const(0x0D)
_SPL06_REG_COEF = const(0x10)
#Parameter selection(sample/sec, times, kT/kP)
_SPL06_PSR_TMP_1 = (0<<4, 0, 524288)
_SPL06_PSR_TMP_2 = (1<<4, 1, 1572864)
_SPL06_PSR_TMP_4 = (2<<4, 2, 3670016)
_SPL06_PSR_TMP_8 = (3<<4, 3, 7864320)
_SPL06_PSR_TMP_16 = (4<<4, 4, 253952)
_SPL06_PSR_TMP_32 = (5<<4, 5, 516096)
_SPL06_PSR_TMP_64 = (6<<4, 6, 1040384)
_SPL06_PSR_TMP_128 = (7<<4, 7, 2088960)
class SPL06:
def __init__(self, i2c_bus, addr=_SPL06_ADDRESS, rate=_SPL06_PSR_TMP_32):
self._device = i2c_bus
self._address = addr
self._rate = rate
self._psr = 0
self._tmp = 0
self._alt = 0
if self._rreg(_SPL06_REG_ID) != 0x10:
raise AttributeError("Cannot find a SPL06-001")
self._init()
def _wreg(self, reg, val):
'''Write memory address'''
self._device.writeto_mem(self._address,reg,val.to_bytes(1, 'little'))
def _rreg(self, reg, nbytes=1):
'''Read memory address'''
return self._device.readfrom_mem(self._address, reg, nbytes)[0] if nbytes<=1 else self._device.readfrom_mem(self._address, reg, nbytes)[0:nbytes]
def _u2s(self, value, n=8):
return value if value < (1 << (n-1)) else value - (1 << n)
def _status(self):
'''数据转换状态'''
status = self._rreg(_SPL06_MEAS_CFG)
return status & 0x80, status & 0x40, (status >> 4 & 0x01) & (status >> 5 & 0x01) #COEF_RDY,SENSOR_RDY,TMP_RDY+PRS_RDY
def _init(self):
'''软复位'''
self._wreg(_SPL06_REG_RST, 0x89)
time.sleep_ms(50)
'''判断校准数据是否就绪，并读取'''
while not self._status()[0]:
time.sleep_ms(1)
buf = self._rreg(_SPL06_REG_COEF, 18)
self._c0 = self._u2s(buf[0] << 4 | buf[1] >> 4, 12)
self._c1 = self._u2s((buf[1] & 0x0F) << 8 | buf[2], 12)
self._c00 = self._u2s(buf[3] << 12 | buf[4] << 4 | buf[5] >> 4, 20)
self._c10 = self._u2s((buf[5] & 0x0F) << 16 | buf[6] << 8 | buf[7], 20)
self._c01 = self._u2s(buf[8] << 8 | buf[9], 16)
self._c11 = self._u2s(buf[10] << 8 | buf[11], 16)
self._c20 = self._u2s(buf[12] << 8 | buf[13], 16)
self._c21 = self._u2s(buf[14] << 8 | buf[15], 16)
self._c30 = self._u2s(buf[16] << 8 | buf[17], 16)
'''判断传感器是否就绪，并设置'''
while not self._status()[1]:
time.sleep_ms(1)
self._wreg(_SPL06_MEAS_CFG, 0x07) #Continuous pressure and temperature
self._wreg(_SPL06_PSR_CFG, self._rate[0] | self._rate[1]) #Configuration of pressure measurement.
self._wreg(_SPL06_TMP_CFG, self._rate[0] | self._rate[1] | 0x80) #Configuration of temperature measurement.
self._rreg(_SPL06_REG_PSR, 6)
if self._rate[1] > 3:
self._wreg(_SPL06_CFG_REG, self._rreg(_SPL06_CFG_REG) | 0x0C) #when the oversampling rate is >8 times.
''''判断数据是否就绪，并读取'''
#while not self._status()[2]:
#time.sleep_ms(1) #数据就绪需要耗时1s左右
@property
def getdata(self):
'''处理获取数据'''
if self._status()[2]:
buf = self._rreg(_SPL06_REG_PSR, 6)
praw = self._u2s(buf[0] << 16 | buf[1] << 8 | buf[2], 24) / self._rate[2]
traw = self._u2s(buf[3] << 16 | buf[4] << 8 | buf[5], 24) / self._rate[2]
try:
self._psr = self._c00 + praw * (self._c10 + praw *(self._c20 + praw * self._c30)) + traw * self._c01 + traw * praw * (self._c11 + praw * self._c21)
except:
self._psr = 0
self._tmp = self._c0 * 0.5 + self._c1 * traw
self._alt = (1 - (self._psr / 101325) ** (1/5.255)) * 44330
return round(self._psr/100, 2), round(self._tmp, 2), round(self._alt,2)
def pressure(self):
return self.getdata[0]
def temperature(self):
return self.getdata[1]
def altitude(self):
return self.getdata[2]
"""
_SPL06-001
MicroPython library for the _SPL06-001(Air pressure sensor)
=======================================================
@dahanzimin From the Mixly Team
"""
import time
from micropython import const
_SPL06_ADDRESS = const(0x77)
_SPL06_REG_PSR = const(0x00)
_SPL06_REG_TMP = const(0x03)
_SPL06_PSR_CFG = const(0x06)
_SPL06_TMP_CFG = const(0x07)
_SPL06_MEAS_CFG = const(0x08)
_SPL06_CFG_REG = const(0x09)
_SPL06_REG_RST = const(0x0C)
_SPL06_REG_ID = const(0x0D)
_SPL06_REG_COEF = const(0x10)
#Parameter selection(sample/sec, times, kT/kP)
_SPL06_PSR_TMP_1 = (0<<4, 0, 524288)
_SPL06_PSR_TMP_2 = (1<<4, 1, 1572864)
_SPL06_PSR_TMP_4 = (2<<4, 2, 3670016)
_SPL06_PSR_TMP_8 = (3<<4, 3, 7864320)
_SPL06_PSR_TMP_16 = (4<<4, 4, 253952)
_SPL06_PSR_TMP_32 = (5<<4, 5, 516096)
_SPL06_PSR_TMP_64 = (6<<4, 6, 1040384)
_SPL06_PSR_TMP_128 = (7<<4, 7, 2088960)
class SPL06:
def __init__(self, i2c_bus, addr=_SPL06_ADDRESS, rate=_SPL06_PSR_TMP_32):
self._device = i2c_bus
self._address = addr
self._rate = rate
self._psr = 0
self._tmp = 0
self._alt = 0
if self._rreg(_SPL06_REG_ID) != 0x10:
raise AttributeError("Cannot find a SPL06-001")
self._init()
def _wreg(self, reg, val):
'''Write memory address'''
self._device.writeto_mem(self._address,reg,val.to_bytes(1, 'little'))
def _rreg(self, reg, nbytes=1):
'''Read memory address'''
return self._device.readfrom_mem(self._address, reg, nbytes)[0] if nbytes<=1 else self._device.readfrom_mem(self._address, reg, nbytes)[0:nbytes]
def _u2s(self, value, n=8):
return value if value < (1 << (n-1)) else value - (1 << n)
def _status(self):
'''数据转换状态'''
status = self._rreg(_SPL06_MEAS_CFG)
return status & 0x80, status & 0x40, (status >> 4 & 0x01) & (status >> 5 & 0x01) #COEF_RDY,SENSOR_RDY,TMP_RDY+PRS_RDY
def _init(self):
'''软复位'''
self._wreg(_SPL06_REG_RST, 0x89)
time.sleep_ms(50)
'''判断校准数据是否就绪，并读取'''
while not self._status()[0]:
time.sleep_ms(1)
buf = self._rreg(_SPL06_REG_COEF, 18)
self._c0 = self._u2s(buf[0] << 4 | buf[1] >> 4, 12)
self._c1 = self._u2s((buf[1] & 0x0F) << 8 | buf[2], 12)
self._c00 = self._u2s(buf[3] << 12 | buf[4] << 4 | buf[5] >> 4, 20)
self._c10 = self._u2s((buf[5] & 0x0F) << 16 | buf[6] << 8 | buf[7], 20)
self._c01 = self._u2s(buf[8] << 8 | buf[9], 16)
self._c11 = self._u2s(buf[10] << 8 | buf[11], 16)
self._c20 = self._u2s(buf[12] << 8 | buf[13], 16)
self._c21 = self._u2s(buf[14] << 8 | buf[15], 16)
self._c30 = self._u2s(buf[16] << 8 | buf[17], 16)
'''判断传感器是否就绪，并设置'''
while not self._status()[1]:
time.sleep_ms(1)
self._wreg(_SPL06_MEAS_CFG, 0x07) #Continuous pressure and temperature
self._wreg(_SPL06_PSR_CFG, self._rate[0] | self._rate[1]) #Configuration of pressure measurement.
self._wreg(_SPL06_TMP_CFG, self._rate[0] | self._rate[1] | 0x80) #Configuration of temperature measurement.
self._rreg(_SPL06_REG_PSR, 6)
if self._rate[1] > 3:
self._wreg(_SPL06_CFG_REG, self._rreg(_SPL06_CFG_REG) | 0x0C) #when the oversampling rate is >8 times.
''''判断数据是否就绪，并读取'''
#while not self._status()[2]:
#time.sleep_ms(1) #数据就绪需要耗时1s左右
@property
def getdata(self):
'''处理获取数据'''
if self._status()[2]:
buf = self._rreg(_SPL06_REG_PSR, 6)
praw = self._u2s(buf[0] << 16 | buf[1] << 8 | buf[2], 24) / self._rate[2]
traw = self._u2s(buf[3] << 16 | buf[4] << 8 | buf[5], 24) / self._rate[2]
try:
self._psr = self._c00 + praw * (self._c10 + praw *(self._c20 + praw * self._c30)) + traw * self._c01 + traw * praw * (self._c11 + praw * self._c21)
except:
self._psr = 0
self._tmp = self._c0 * 0.5 + self._c1 * traw
self._alt = (1 - (self._psr / 101325) ** (1/5.255)) * 44330
return round(self._psr/100, 2), round(self._tmp, 2), round(self._alt,2)
def pressure(self):
return self.getdata[0]
def temperature(self):
return self.getdata[1]
def altitude(self):
return self.getdata[2]