[openeuler@tc ~]$ pip install paho-mqtt -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com
[openeuler@tc ~]$ pip install aiocoap -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com
[openeuler@tc ~]$ pip install modbus -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com
[openeuler@tc ~]$ pip install pymodbus  -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com
[openeuler@tc ~]$ pip install pyserial  -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com

import paho.mqtt.client as mqtt
import time

"""
　　創建了一個 MQTT 客戶端，連接到指定的 MQTT 代理服務器，并訂閱了一個主題（"example_topic"）。
　　然后，它進入一個循環，等待用戶輸入消息并將其發布到指定的主題。
　　同時，它也監聽來自訂閱主題的消息，并在收到消息時調用 on_message 函數來處理消息。
　　這里需要將 broker_address 更改為實際的 MQTT 代理服務器地址，并根據你的需求更改主題名稱和其他設置。此示例用于演示基本的 MQTT 消息發布和訂閱功能。
"""


# MQTT 代理服務器的地址和端口
broker_address = "mqtt.eclipse.org"
port = 1883

# 創建 MQTT 客戶端
client = mqtt.Client("Python_Client")

# 連接到 MQTT 代理服務器
client.connect(broker_address, port)

# 訂閱一個主題
def on_message(client, userdata, message):
    print(f"Received message '{message.payload}' on topic '{message.topic}'")

client.on_message = on_message
client.subscribe("example_topic")

# 啟動 MQTT 客戶端的循環
client.loop_start()

# 發布消息
while True:
    message = input("Enter a message to publish: ")
    client.publish("example_topic", message)
    time.sleep(1)

# 程序結束時斷開連接
client.disconnect()

import logging
import asyncio
import aiocoap

"""
    安裝包：pip install aiocoap
    CoAP服務器提供一個資源 "hello"，用于響應 GET 請求，并返回 "Hello, CoAP World!"。
    CoAP（Constrained Application Protocol）是一種用于受限環境的輕量級應用層協議，通常用于物聯網（IoT）設備之間的通信。
    CoAP客戶端構建一個 GET 請求，請求服務器上的資源。確保將服務器示例中的 coap://localhost/hello 替換為你的服務器地址和資源路徑。
    CoAP是一種輕量級協議，適用于受限環境，適合IoT設備和嵌入式系統。
"""

# 設置日志級別以查看詳細信息
logging.basicConfig(level=logging.INFO)

async def coap_client():
    protocol = await aiocoap.Context.create_client_context()

    # 構建CoAP請求
    request = aiocoap.Message(code=aiocoap.GET)
    request.set_request_uri('coap://localhost/hello')

    try:
        response = await protocol.request(request).response
        print("Response: %s" % response.payload.decode('utf-8'))
    except Exception as e:
        print("Failed to fetch resource:")
        print(e)

if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(coap_client())

import logging
import asyncio
import aiocoap.resource as resource
import aiocoap

# 設置日志級別以查看詳細信息
logging.basicConfig(level=logging.INFO)

# 定義一個CoAP資源
class HelloWorldResource(resource.Resource):
    def __init__(self):
        super(HelloWorldResource, self).__init__()

    async def render_get(self, request):
        return aiocoap.Message(payload=b"Hello, CoAP World!")

# 創建CoAP服務器
root = resource.Site()
root.add_resource(('hello',), HelloWorldResource())

# 運行服務器
def main():
    asyncio.Task(aiocoap.Context.create_server_context(root))
    asyncio.get_event_loop().run_forever()

if __name__ == "__main__":
    main()

from pymodbus.client.sync import ModbusTcpClient

"""
    Modbus是一種用于工業自動化和通信的通信協議，它被廣泛用于監控和控制設備。
    以下是一個簡單的Python Modbus示例，使用pymodbus庫來與Modbus設備通信。
    在這個示例中，我們將使用Modbus TCP通信，它是一種常見的Modbus通信方式。
    Modbus客戶端連接到Modbus服務器，讀取保持寄存器中的數據，然后斷開連接。
    Modbus服務器創建一個數據存儲塊，該塊包含保持寄存器中的數據，并在指定的IP地址和端口上啟動Modbus TCP服務器。
    請注意，這只是一個非常簡單的示例，你可以根據你的具體需求和Modbus設備的特點來進行更復雜的通信和數據處理。
    Modbus有多個變種和數據類型，你需要根據你的設備和協議規范進行適當的設置。
"""

# Modbus服務器的IP地址和端口
server_ip = '127.0.0.1'
server_port = 502

# 創建Modbus TCP客戶端
client = ModbusTcpClient(server_ip, port=server_port)

# 連接到Modbus服務器
client.connect()

# 讀取保持寄存器中的數據
result = client.read_holding_registers(0, 2, unit=1)  # 從單元ID 1中讀取前兩個保持寄存器

if not result.isError():
    print(f'Read data: {result.registers}')
else:
    print(f'Error reading data: {result.message}')

# 斷開連接
client.close()

from pymodbus.server.sync import StartTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock
from pymodbus.datastore import ModbusServerContext

# 創建Modbus服務器上的數據存儲塊
store = ModbusServerContext(slaves=1, single=True)
block = ModbusSequentialDataBlock(0, [0, 1, 2, 3], 0, [0]*4)
store.register(block, 0)

# 啟動Modbus TCP服務器
StartTcpServer(store, address=('127.0.0.1', 502))

import serial
import time

"""
    安裝包：pip install pyserial
    Python中與RS-485通信的示例需要使用適當的硬件支持（例如RS-485適配器）以及串口通信庫，通常使用pyserial庫進行RS-485通信。
    以下是一個簡單的Python RS-485示例，假設你已經配置好了RS-485硬件。
    在這個示例中，我們使用pyserial庫配置了RS-485通信參數，包括串口設備路徑、波特率、校驗位、停止位、數據位和超時時間。
    然后，我們發送了一個字符串數據并嘗試讀取從RS-485設備返回的響應數據。
    確保將port設置為你的RS-485串口設備的路徑（通常是/dev/ttyUSB0或/dev/ttyS0等，具體取決于你的系統）。
    此示例僅用于演示RS-485通信的基本原理，你可以根據你的應用需求進行更復雜的通信和數據處理。
"""

# 配置串口參數
ser = serial.Serial(
    port='/dev/ttyUSB0',  # 串口設備路徑，根據實際情況修改
    baudrate=9600,        # 波特率
    parity=serial.PARITY_NONE,
    stopbits=serial.STOPBITS_ONE,
    bytesize=serial.EIGHTBITS,
    timeout=1            # 超時時間（秒）
)

try:
    # 發送數據
    ser.write(b'Hello, RS-485!')

    # 讀取數據
    response = ser.readline().decode('utf-8').strip()
    print(f'Received: {response}')

except Exception as e:
    print(f'Error: {e}')

finally:
    # 關閉串口連接
    ser.close()

import socket

"""
    TCP/IP 服務器等待客戶端連接，接收來自客戶端的數據，發送響應數據，然后關閉客戶端連接。
    TCP協議提供了可靠的、面向連接的通信，適用于需要穩定數據傳輸的應用程序。
    確保在服務器和客戶端示例中的 server_address 中設置正確的IP地址和端口號，以便它們能夠相互通信。
    這只是一個簡單的示例，你可以根據需要構建更復雜的 TCP/IP 應用程序
"""

# 創建TCP客戶端套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 服務器地址和端口
server_address = ('127.0.0.1', 12345)

# 連接到服務器
client_socket.connect(server_address)

# 發送數據給服務器
message = "Hello, TCP Server!"
client_socket.send(message.encode('utf-8'))

# 接收服務器的響應
response = client_socket.recv(1024)
print(f"Received response: {response.decode('utf-8')}")

# 關閉客戶端連接
client_socket.close()

import socket

# 創建TCP服務器套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 綁定服務器地址和端口
server_address = ('127.0.0.1', 12345)
server_socket.bind(server_address)

# 監聽客戶端連接
server_socket.listen(5)
print("TCP Server is listening...")

while True:
    # 等待客戶端連接
    client_socket, client_address = server_socket.accept()
    print(f"Connected to client: {client_address}")

    # 接收客戶端發送的數據
    data = client_socket.recv(1024)
    if data:
        print(f"Received data: {data.decode('utf-8')}")

    # 發送響應數據給客戶端
    response = "Hello, TCP Client!"
    client_socket.send(response.encode('utf-8'))

    # 關閉客戶端連接
    client_socket.close()

import socket

"""
    Python UDP（User Datagram Protocol）示例，其中一個程序作為UDP服務器接收消息，另一個程序作為UDP客戶端發送消息。
    UDP是一種面向無連接的協議，通常用于實現簡單的消息傳遞。
    UDP是一種無連接協議，不提供可靠的數據傳輸，也不保證消息的順序。
    因此，在實際應用中，你可能需要添加錯誤檢查和恢復機制來處理丟失的消息或消息亂序的情況，具體取決于你的需求。
"""

# 創建UDP套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 服務器地址和端口
server_address = ('localhost', 12345)

while True:
    message = input("Enter a message to send to the server: ")
    if not message:
        break

    # 發送消息到服務器
    client_socket.sendto(message.encode(), server_address)

# 關閉套接字
client_socket.close()

import socket

# 創建UDP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 綁定服務器地址和端口
server_address = ('localhost', 12345)
server_socket.bind(server_address)

print("UDP Server is listening...")

while True:
    # 接收來自客戶端的數據
    data, client_address = server_socket.recvfrom(1024)
    print(f"Received message: {data.decode()} from {client_address}")

# 關閉套接字
server_socket.close()

import requests

"""
    在Python中，你可以使用內置的HTTP庫 http.server 來創建一個簡單的HTTP服務器，或使用第三方庫 requests 來創建HTTP客戶端。
    在這個示例中，我們使用第三方庫 requests 發起一個HTTP GET請求，然后檢查響應的狀態碼和解析響應數據。
    此示例中，我們使用一個公共的 JSONPlaceholder API 作為示例，你可以根據自己的需求修改URL。
"""

# 發起HTTP GET請求
url = "https://jsonplaceholder.typicode.com/posts/1"
response = requests.get(url)

# 檢查響應狀態碼
if response.status_code == 200:
    # 解析響應數據
    data = response.json()
    print("Response data:")
    print(data)
else:
    print(f"Failed to retrieve data. Status code: {response.status_code}")

# 導入HTTP服務器模塊
import http.server
import socketserver

# 指定端口號
port = 8000

# 創建HTTP服務器
with socketserver.TCPServer(("", port), http.server.SimpleHTTPRequestHandler) as httpd:
    print(f"HTTP server is running on port {port}")
    httpd.serve_forever()

import requests

"""
    在Python中創建一個HTTPS服務器和使用HTTPS客戶端，你需要使用SSL/TLS協議來加密通信。分別使用內置庫和第三方庫requests
    我們創建一個HTTPS服務器并配置SSL證書和密鑰文件（server.pem）。你需要自己生成或獲取有效的SSL證書和密鑰文件。
    在HTTPS客戶端示例中，我們使用 requests 庫發送一個HTTPS GET請求，同時提供了服務器的SSL證書文件（server.pem）來驗證服務器的身份。
    請確保將示例URL替換為你要訪問的實際HTTPS網站。
    這只是一個簡單的HTTPS服務器和HTTPS客戶端示例。
    要在生產環境中使用HTTPS，你需要獲取有效的SSL證書，通常從信任的證書頒發機構（CA）購買或申請。
    不要在生產環境中使用自簽名證書（像上面示例中的server.pem），因為瀏覽器和應用程序通常不會信任它們。
"""

# 發起HTTPS GET請求
url = "https://example.com"  # 請將示例URL替換為實際的HTTPS網站
response = requests.get(url, verify="server.pem")

# 檢查響應狀態碼
if response.status_code == 200:
    # 處理響應數據
    data = response.text
    print("Response data:")
    print(data)
else:
    print(f"Failed to retrieve data. Status code: {response.status_code}")

import http.server
import socketserver
import ssl



# 指定端口號
port = 8443

# 創建HTTP服務器
with socketserver.TCPServer(("", port), http.server.SimpleHTTPRequestHandler) as httpd:
    # 配置SSL證書和密鑰
    httpd.socket = ssl.wrap_socket(httpd.socket, certfile="server.pem", server_side=True)

    print(f"HTTPS server is running on port {port}")
    httpd.serve_forever()