|
|
本帖最后由 穹顶之下2018 于 2025-12-24 19:43 编辑
摘要
本文介绍socket编程在实际生活中的常见应用。本文需要读者具有计算机网络基础知识（http，udp，tcp等）、Python编程基础知识。
注：代码只需微改即可运行在pc端。下文demo均运行在esp32s3上，并测试通过。其中涉及的账号密码问题，需读者自行注册或修改。
心知天气【需私钥】、百度热搜【公开api】、网络NTP时钟
简要说明：
心知和百度，都是发起http get请求，返回json格式的结果，而后解析获取相应数据。
NTP网络时钟则是向指定ntp服务器的特定端口发起udp请求，而后解析返回的数据。
测试demo
#----------------------------------------------------------
import network,time
wlan = network.WLAN()
wlan.active(True)
def do_connect(ssid,passwd):
if not wlan.isconnected(): #wlan.scan()
print('connecting to network...')
wlan.connect(ssid,passwd)
timout = 0
while not wlan.isconnected():
time.sleep(1)
if timout >= 10:
break
print('network config:', wlan.ipconfig('addr4'))
print('MAC config:', ''.join(['{:02X}'.format(x) for x in wlan.config('mac')]))
do_connect('你的wifi账号','你的wifi密码')
import json,requests
#百度热搜的JSON接口
def getBaiduHotInfo(req):
topic = ['homepage','realtime','livelihood','finance','phrase','novel','movie','teleplay','car','game']
url = 'https://top.baidu.com/api/board?platform=wise&tab=' #完整链接示意：'https://top.baidu.com/api/board?platform=wise&tab=realtime'
baiduhoturl1 = "https://top.baidu.com/board?tab=realtime" # 百度热搜网页版官方 API
baiduhoturl2= "https://api.1314.cool/getbaiduhot/" #返回百度实时热点共30条，每3分钟更新一次
try:
res = requests.get(url+topic[req])
except Exception as err:
print("getBaiduHotInfo failed...", err)
return None
else:
res = res.content
_res = json.loads(res)
words = []
if _res['success'] is True and _res['error']['code'] == 0:
ptr = _res['data']['cards'][0]['content'][0]['content']
for word in ptr:
if word['isTop'] is False:
print(word['index'],word['word']) #debug
words.append(word['word'])
#print(words) #debug
return words
private_key = '你的私钥，注册心知获得'
public_key = ""
city = 'shenzhen'
def getWeatherInfo(req):
global city
weatherapi1 = "https://weathernew.pae.baidu.com/weathernew/pc?query=%E5%B9%BF%E4%B8%9C%E6%B7%B1%E5%9C%B3%E5%A4%A9%E6%B0%94&srcid=4982&forecast=long_day_forecast"
'''
心知 在线文档 https://seniverse.yuque.com/hyper_data/datasets/start?
加密传输 https://seniverse.yuque.com/hyper_data/api_v4/nke5p8
'''
apidict = { #k必须是不可变的，如字符串，数字或元组 v可以取任何数据类型
#天气实况
'weather':'https://api.seniverse.com/v3/weather/now.json?key=your_api_key&location=city&language=zh-Hans&unit=c',
#生活指数
'life':'https://api.seniverse.com/v3/life/suggestion.json?key=your_api_key&location=city&language=zh-Hans&days=5',
#获取指定城市未来最多 15 天每天的白天和夜间预报，以及昨日的历史天气。付费用户可获取全部数据，免费用户只返回3天天气预报
'daily':'https://api.seniverse.com/v3/weather/daily.json?key=your_api_key&location=city&language=zh-Hans&unit=c&start=0&days=5',
#农历、节气、生肖
'calendar':'https://api.seniverse.com/v3/life/chinese_calendar.json?key=your_api_key&start=0&days=7'
}
def getFinalAPI(req):
cmdList = ('weather','life','daily','calendar')
tmp = apidict[cmdList[req]]
if tmp.find('your_api_key') > -1 and tmp.find('city') > -1:
url = tmp.replace("your_api_key",private_key,1).replace("city",city,1)
elif tmp.find('your_api_key') > -1:
url = tmp.replace("your_api_key",private_key,1)
print(url) #debug
return url
try:
res = requests.get(getFinalAPI(req))
except Exception as err:
print("getWeatherInfo failed...", err)
return None
else:
res = res.content #print(res) debug
_res = json.loads(res)
if req == 0: #TODO
print(_res['results'][0]['now'])
elif req == 1:
pass #内容太多，TODO
elif req == 2: #TODO
for daily in _res['results'][0]['daily']:
print(daily['low'],'~',daily['high'])
elif req == 3:
pass #b'{"status":"You do not have access to this API.","status_code":"AP010002"}'
import ntptime
#https://github.com/micropython/micropython-lib/blob/master/micropython/net/ntptime/ntptime.py
#https://docs.micropython.org/en/ ... html#time.localtime
def getBeijingTimeInfo():
ALIBABA_NTP_SERVERS = [ # 阿里巴巴 NTP 服务器列表 库默认pool.ntp.org
"ntp.aliyun.com", #主服务器
"ntp1.aliyun.com", #备用服务器1
"ntp2.aliyun.com", #备用服务器2
"ntp3.aliyun.com" #备用服务器3
]
ntptime.host = ALIBABA_NTP_SERVERS[0] #优先使用主服务器
ntptime.timeout = 0.1 #NTP请求超时时间
#ntptime.settime() #同步时间
#循环尝试备用服务器（增强稳定性）
'''
for server in ALIBABA_NTP_SERVERS:
try:
print(f"Trying NTP server: {server}")
ntptime.host = server
ntptime.settime()
print(f"Successfully synced with {server}")
return True
except OSError as e:
print(f"Failed to sync with {server}: {e}")
time.sleep(1) # 失败后延迟1秒重试
print("All NTP servers failed!")
return False
'''
try:
timestamp = ntptime.time()
#if time.gmtime(0)[0]==2000:
# timestamp += (3155673600-2208988800) #转回标准的1970开始计时
timestamp += (8*3600) #东八区
'''
month is 1-12
mday is 1-31
hour is 0-23
minute is 0-59
second is 0-59
weekday is 0-6 for Mon-Sun
yearday is 1-366
'''
timinfo = time.localtime(timestamp) #秒数转时间
print(timinfo)
return timinfo
except Exception as err:
print('get nettime err...',err)
return None
getBaiduHotInfo(1) #获取百度热搜json结果
getWeatherInfo(0) #获取心知天气json结果
getBeijingTimeInfo() #获取ntp服务器的返回结果
#----------------------------------------------------------
网页返回内容截图
热搜结果
天气结果
{"results":[{"location":{"id":"WS10730EM8EV","name":"深圳","country":"CN","path":"深圳,深圳,广东,中国","timezone":"Asia/Shanghai","timezone_offset":"+08:00"},"now":{"text":"晴","code":"1","temperature":"20"},"last_update":"2025-12-24T19:20:22+08:00"}]}
NTP实现
后续内容预告
后续将介绍除hls协议之外，当前网络fm项目最最常用的 Icecast Stream 直播协议【超级简单socket连接上即可一直读音频流，直接送入解码即可】。将以http mp3 fm音频流为例进行分析与实战。
如果读者有更有意思的适合mcu的网络应用，可以说说。
|
-
|
发表于 2025-12-24 19:36:39
