① Android+python數據傳輸
實現Android與Python數據傳輸的目標是為了讓手機操控機器人,而機器人控制與視覺運用Python語言。此操作簡化了機器人控制與交互。本文記錄了如何實現基本通訊。
在Python端,首先需設置埠號,主機IP可自行設定。主機IP需通過socket庫函數獲取。隨後創建socket。在socket創建後,進入while循環,形成完整代碼。
Android端的主要步驟包括三步:設置埠號、主機IP,創建socket,運行代碼。完整代碼如下。在運行過程中,可能遇到Java.net.SocketException: socket failed: EPERM (Operation not permitted)的錯誤提示。這表明Android程序的網路許可權未開啟。需在AndroidManifest.xml中添加網路許可權。
最終效果:單擊一次按鈕實現一次數據傳輸。
② 使用樹莓派和Python實現ModbusTCP通訊
要在樹莓派上使用Python實現ModbusTCP通訊,可以按照以下步驟進行:
1. 構建系統環境 在Windows 10的台式機上安裝Anaconda發行版的Python 3.6。 樹莓派上使用官方Raspberry Pi系統,並安裝Python 3.7。 使用pip3命令在樹莓派的Python環境中安裝modbus_tk庫。
2. 配置區域網環境 確保樹莓派和台式機位於同一區域網內。 設定樹莓派的IP地址和用於ModbusTCP通訊的開放埠。 在樹莓派上安裝並配置ufw防火牆,打開指定的ModbusTCP通訊埠。
3. 編寫ModbusTCP通訊程序 樹莓派從機程序:根據modbus_tk庫的API編寫,用於響應來自Modbus主機的數據請求。 台式機主機程序:同樣基於modbus_tk庫編寫,用於發送數據請求並接收來自樹莓派從機的響應。
4. 整合Modbus功能代碼 將Modbus協議的相關參數和結構定義在一個文件中,便於從機和主機程序共同引用。
5. 運行與測試 在樹莓派上運行從機程序,確保其監聽指定的ModbusTCP埠。 在台式機上運行主機程序,發送數據請求並觀察從機響應。 根據需要進行調試和優化,確保通訊的穩定性和數據的准確性。
通過以上步驟,您可以在樹莓派與電腦之間實現ModbusTCP通訊,從而進行設備數據的讀取和發送。
③ python怎麼建立socket服務端
socket伺服器再細分可分為多種了,tcp,udp,websocket,都是調用socket模塊,但是具體實現起來有一點細微的差別
先給出一個tcp和udp通過socket協議實現的聊天室的例子
python聊天室(python2.7版本):
都是分別運行server.py和client.py,就可以進行通訊了。
TCP版本:
socket-tcp-server.py(服務端):
#-*-encoding:utf-8-*-
#socket.getaddrinfo(host,port,family=0,socktype=0,proto=0,flags=0)
#根據給定的參數host/port,相應的轉換成一個包含用於創建socket對象的五元組,
#參數host為域名,以字元串形式給出代表一個IPV4/IPV6地址或者None.
#參數port如果字元串形式就代表一個服務名,比如「http」"ftp""email"等,或者為數字,或者為None
#參數family為地主族,可以為AF_INET,AF_INET6,AF_UNIX.
#參數socktype可以為SOCK_STREAM(TCP)或者SOCK_DGRAM(UDP)
#參數proto通常為0可以直接忽略
#參數flags為AI_*的組合,比如AI_NUMERICHOST,它會影響函數的返回值
#附註:給參數host,port傳遞None時建立在C基礎,通過傳遞NULL。
#該函數返回一個五元組(family,socktype,proto,canonname,sockaddr),同時第五個參數sockaddr也是一個二元組(address,port)
#更多的方法及鏈接請訪問
#Echoserverprogram
fromsocketimport*
importsys
importthreading
fromtimeimportctime
fromtimeimportlocaltime
importtraceback
importtime
importsubprocess
reload(sys)
sys.setdefaultencoding("utf8")
HOST='127.0.0.1'
PORT=8555#設置偵聽埠
BUFSIZ=1024
classTcpServer():
def__init__(self):
self.ADDR=(HOST,PORT)
try:
self.sock=socket(AF_INET,SOCK_STREAM)
print'%disopen'%PORT
self.sock.bind(self.ADDR)
self.sock.listen(5)
#設置退出條件
self.STOP_CHAT=False
#所有監聽的客戶端
self.clients={}
self.thrs={}
self.stops=[]
exceptException,e:
print"%disdown"%PORT
returnFalse
defIsOpen(ip,port):
s=socket(AF_INET,SOCK_STREAM)
try:
s.connect((ip,int(port)))
#s.shutdown(2)
#利用shutdown()函數使socket雙向數據傳輸變為單向數據傳輸。shutdown()需要一個單獨的參數,
#該參數表示s了如何關閉socket。具體為:0表示禁止將來讀;1表示禁止將來寫;2表示禁止將來讀和寫。
print'%disopen'%port
returnTrue
except:
print'%disdown'%port
returnFalse
deflisten_client(self):
whilenotself.STOP_CHAT:
print(u'等待接入,偵聽埠:%d'%(PORT))
self.tcpClientSock,self.addr=self.sock.accept()
print(u'接受連接,客戶端地址:',self.addr)
address=self.addr
#將建立的clientsocket鏈接放到列表self.clients中
self.clients[address]=self.tcpClientSock
#分別將每個建立的鏈接放入進程中,接收且分發消息
self.thrs[address]=threading.Thread(target=self.readmsg,args=[address])
self.thrs[address].start()
time.sleep(0.5)defreadmsg(self,address):
#如果地址不存在,則返回False
ifaddressnotinself.clients:
returnFalse
#得到發送消息的clientsocket
client=self.clients[address]
whileTrue:
try:
#獲取到消息內容data
data=client.recv(BUFSIZ)
except:
print(e)
self.close_client(address)
break
ifnotdata:
break
#python3使用bytes,所以要進行編碼
#s='%s發送給我的信息是:[%s]%s'%(addr[0],ctime(),data.decode('utf8'))
#對日期進行一下格式化
ISOTIMEFORMAT='%Y-%m-%d%X'
stime=time.strftime(ISOTIMEFORMAT,localtime())
s=u'%s發送給我的信息是:%s'%(str(address),data.decode('utf8'))
#將獲得的消息分發給鏈接中的clientsocket
forkinself.clients:
self.clients[k].send(s.encode('utf8'))
self.clients[k].sendall('sendall:'+s.encode('utf8'))
printstr(k)
print([stime],':',data.decode('utf8'))
#如果輸入quit(忽略大小寫),則程序退出
STOP_CHAT=(data.decode('utf8').upper()=="QUIT")
ifSTOP_CHAT:
print"quit"
self.close_client(address)
print"alreadyquit"
break
defclose_client(self,address):
try:
client=self.clients.pop(address)
self.stops.append(address)
client.close()
forkinself.clients:
self.clients[k].send(str(address)+u"已經離開了")
except:
pass
print(str(address)+u'已經退出')
if__name__=='__main__':
tserver=TcpServer()
tserver.listen_client()
——————————華麗的分割線——————————
socket-tcp-client.py(客戶端):
#-*-encoding:utf-8-*-
fromsocketimport*
importsys
importthreading
importtime
reload(sys)
sys.setdefaultencoding("utf8")
#測試,連接本機
HOST='127.0.0.1'
#設置偵聽埠
PORT=8555
BUFSIZ=1024
classTcpClient:
ADDR=(HOST,PORT)
def__init__(self):
self.HOST=HOST
self.PORT=PORT
self.BUFSIZ=BUFSIZ
#創建socket連接
self.client=socket(AF_INET,SOCK_STREAM)
self.client.connect(self.ADDR)
#起一個線程,監聽接收的信息
self.trecv=threading.Thread(target=self.recvmsg)
self.trecv.start()
defsendmsg(self):
#循環發送聊天消息,如果socket連接存在則一直循環,發送quit時關閉鏈接
whileself.client.connect_ex(self.ADDR):
data=raw_input('>:')
ifnotdata:
break
self.client.send(data.encode('utf8'))
print(u'發送信息到%s:%s'%(self.HOST,data))
ifdata.upper()=="QUIT":
self.client.close()
printu"已關閉"
break
defrecvmsg(self):
#接收消息,如果鏈接一直存在,則持續監聽接收消息
try:
whileself.client.connect_ex(self.ADDR):
data=self.client.recv(self.BUFSIZ)
print(u'從%s收到信息:%s'%(self.HOST,data.decode('utf8')))
exceptException,e:
printstr(e)
if__name__=='__main__':
client=TcpClient()
client.sendmsg()
UDP版本:
socket-udp-server.py
#-*-coding:utf8-*-
importsys
importtime
importtraceback
importthreading
reload(sys)
sys.setdefaultencoding('utf-8')
importsocket
importtraceback
HOST="127.0.0.1"
PORT=9555
CHECK_PERIOD=20
CHECK_TIMEOUT=15
classUdpServer(object):
def__init__(self):
self.clients=[]
self.beats={}
self.ADDR=(HOST,PORT)
try:
self.sock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
self.sock.bind(self.ADDR)#綁定同一個域名下的所有機器
self.beattrs=threading.Thread(target=self.checkheartbeat)
self.beattrs.start()
exceptException,e:
traceback.print_exc()
returnFalse
deflisten_client(self):
whileTrue:
time.sleep(0.5)
print"hohohohohoo"
try:
recvData,address=self.sock.recvfrom(2048)
ifnotrecvData:
self.close_client(address)
break
ifaddressinself.clients:
senddata=u"%s發送給我的信息是:%s"%(str(address),recvData.decode('utf8'))
ifrecvData.upper()=="QUIT":
self.close_client(address)
ifrecvData=="HEARTBEAT":
self.heartbeat(address)
continue
else:
self.clients.append(address)
senddata=u"%s發送給我的信息是:%s"%(str(address),u'進入了聊天室')
forcinself.clients:
try:
self.sock.sendto(senddata,c)
exceptException,e:
printstr(e)
self.close_client(c)
exceptException,e:
#traceback.print_exc()
printstr(e)
pass
defheartbeat(self,address):
self.beats[address]=time.time()
defcheckheartbeat(self):
whileTrue:
print"checkheartbeat"
printself.beats
try:
forcinself.clients:
printtime.time()
printself.beats[c]
ifself.beats[c]+CHECK_TIMEOUT<time.time():
printu"%s心跳超時,連接已經斷開"%str(c)
self.close_client(c)
else:
printu"checkp%s,沒有斷開"%str(c)
exceptException,e:
traceback.print_exc()
printstr(e)
pass
time.sleep(CHECK_PERIOD)
defclose_client(self,address):
try:
ifaddressinself.clients:
self.clients.remove(address)
ifself.beats.has_key(address):
delself.beats[address]
printself.clients
forcinself.clients:
self.sock.sendto(u'%s已經離開了'%str(address),c)
print(str(address)+u'已經退出')
exceptException,e:
printstr(e)
raise
if__name__=="__main__":
udpServer=UdpServer()
udpServer.listen_client()
——————————華麗的分割線——————————
socket-udp-client.py:
#-*-coding:utf8-*-
importsys
importthreading
importtime
reload(sys)
sys.setdefaultencoding('utf-8')
importsocket
HOST="127.0.0.1"
PORT=9555
#BEAT_PORT=43278
BEAT_PERIOD=5
classUdpClient(object):
def__init__(self):
self.clientsock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
self.HOST=HOST
self.ADDR=(HOST,PORT)
self.clientsock.sendto(u'請求建立鏈接',self.ADDR)
self.recvtrs=threading.Thread(target=self.recvmsg)
self.recvtrs.start()
self.hearttrs=threading.Thread(target=self.heartbeat)
self.hearttrs.start()
defsendmsg(self):
whileTrue:
data=raw_input(">:")
ifnotdata:
break
self.clientsock.sendto(data.encode('utf-8'),self.ADDR)
ifdata.upper()=='QUIT':
self.clientsock.close()
break
defheartbeat(self):
whileTrue:
self.clientsock.sendto('HEARTBEAT',self.ADDR)
time.sleep(BEAT_PERIOD)
defrecvmsg(self):
whileTrue:
recvData,addr=self.clientsock.recvfrom(1024)
ifnotrecvData:
break
print(u'從%s收到信息:%s'%(self.HOST,recvData.decode('utf8')))if__name__=="__main__":
udpClient=UdpClient()
udpClient.sendmsg()
④ Python:小白花了兩個小時干出來的python 版通訊錄(大神勿噴)
Python通訊錄程序是一個基於字典實現的基本通訊錄管理系統,具備增刪改查功能。以下是該程序的主要特點和功能模塊:
主要功能:
- 添加聯系人:用戶可以輸入聯系人的姓名、電話等信息,程序會將這些信息存儲到字典中,使用編號作為鍵來唯一標識每個聯系人。
- 刪除聯系人:用戶輸入要刪除的聯系人的編號,程序會根據編號從字典中刪除對應的聯系人信息。
- 修改聯系人信息:用戶輸入要修改的聯系人的編號,然後輸入新的信息,程序會更新字典中對應聯系人的信息。
- 查找聯系人:用戶輸入聯系人的編號,程序會顯示對應聯系人的全部信息。
技術特點:
- 字典存儲:使用Python中的字典數據結構來存儲聯系人信息,通過編號作為鍵,實現了高效的數據管理和訪問。
- 異常處理:在代碼中加入了異常處理機制,以處理用戶輸入錯誤或程序運行中的其他異常情況,提高了程序的健壯性。
- 文件操作:程序可能支持將聯系人信息保存到文件中,以便在程序關閉後仍能保留數據。同時,也可以從文件中讀取聯系人信息,實現數據的持久化存儲。
- 多線程:雖然具體實現細節未給出,但多線程技術可以用於提高程序的響應速度,特別是在處理大量數據或進行復雜操作時。
用戶操作流程:
- 用戶可以通過簡單的命令行界面與程序進行交互。
- 根據提示輸入相應的命令和數字來選擇要執行的操作。
- 根據操作要求輸入相應的信息,如聯系人姓名、電話等。
該Python通訊錄程序是一個很好的自學Python的練習項目,不僅涉及了基本的編程知識,還鍛煉了問題解決能力和代碼實現能力。對於初學者來說,這是一個很好的學習資源和參考示例。