网络数据收发

image.png
功能需求：

  • TCP客户端模式：
    • 连接指定服务器
    • 接受服务器的消息
    • 给服务器发消息
  • TCP服务器模式：
    • 创建服务器
    • 接受客户端消息
    • 回复客户端消息
  • UDP模式：
    • 绑定本地IP和端口
    • 向目标IP和端口发消息
  • 清空消息记录

串口数据收发

依赖库：pyserial
源码：https://github.com/pyserial/pyserial
安装依赖：

  1. pip install pyserial

image.png
功能需求：

  • 扫描串口设备
  • 进行串口连接配置
  • 连接设备
  • 接收串口消息
  • 发送串口消息

使用**pyserial**封装的串口驱动：

  1. import serial
  2. from serial import Serial
  3. from serial.tools import list_ports
  6. def scan_serial_ports():
  7.     ports = list_ports.comports()
  8.     # for port in serial_ports: # ListPortInfo
  9.     #     print(port.device, port.description, port.hwid, port.vid, port.pid, port.serial_number, port.location)
  10.     return [(port.device, port.description) for port in ports]
  13. class SerialDevice:
  14.     def __init__(self, port, baud_rate=9600, timeout=None):
  15.         self.port = port
  16.         self.baud_rate = baud_rate
  17.         self.timeout = timeout
  18.         self.serial: Serial = None
  20.     def open(self):
  21.         try:
  22.             self.serial = serial.Serial(self.port, self.baud_rate, timeout=self.timeout)
  23.             if self.serial.is_open:
  24.                 # print(f"Serial port {self.port} opened successfully.")
  25.                 return True, f"Serial port {self.port} opened successfully."
  26.         except serial.SerialException as e:
  27.             print("请检查设备是否连接，或端口被其他软件占用，Failed to open serial port:", str(e))
  28.             return False, str(e)
  30.         return False, f"Serial port {self.port} open failed."
  32.     def close(self):
  33.         if self.serial and self.serial.is_open:
  34.             self.serial.close()
  35.             print("Serial port closed.")
  36.         else:
  37.             print("Serial port is not open.")
  39.     def write(self, data):
  40.         if not self.serial or not self.serial.is_open:
  41.             print("Serial port is not open.")
  42.             return None
  44.         try:
  45.             self.serial.write(data)
  46.             print("Data written:", data)
  47.         except serial.SerialException as e:
  48.             print("Failed to write data:", str(e))
  50.     def flush(self):
  51.         if not self.serial or not self.serial.is_open:
  52.             print("Serial port is not open.")
  53.             return None
  55.         self.serial.flush()
  57.     def read(self, num_bytes):
  58.         if not self.serial or not self.serial.is_open:
  59.             print("Serial port is not open.")
  60.             return None
  61.         try:
  62.             return self.serial.read(num_bytes)
  63.         except serial.SerialException as e:
  64.             print("Failed to read data:", str(e))
  66.         return None
  68.     def readline(self):
  69.         if not self.serial or not self.serial.is_open:
  70.             print("Serial port is not open.")
  71.             return
  73.         try:
  74.             return self.serial.readline()
  75.         except serial.SerialException as e:
  76.             print("Failed to read data:", str(e))
  78.         return None
  80.     def is_open(self):
  81.         if not self.serial:
  82.             return False
  84.         return self.serial.is_open
  87. if __name__ == '__main__':
  89.     # 示例用法
  90.     serial_ports = scan_serial_ports()
  91.     if len(serial_ports) > 0:
  92.         print("Available serial serial_ports:")
  93.         for device, description in serial_ports:
  94.             print(device, "---", description)
  95.     else:
  96.         print("No serial serial_ports found.")
  98.     # 示例用法
  99.     # sp = SerialDevice("COM1", baud_rate=9600, timeout=1)  # 替换为您的串口名称、波特率和超时时间
  100.     # sp.open()
  101.     # sp.write(b"Hello, Serial!")  # 发送数据
  102.     # print(sp.read(10))  # 读取10个字节的数据
  103.     # sp.close()

蓝牙数据收发

依赖库：pybluez2==0.37
https://github.com/pybluez/pybluez
安装依赖：

  1. pip install pybluez2==0.37

image.png
功能需求：

  • 扫描蓝牙设备
  • 连接蓝牙设备
  • 接受蓝牙数据
  • 发送蓝牙数据

利用**pybluez**封装的蓝牙驱动：

  1. import bluetooth
  4. # 扫描所有设备
  5. class BluetoothDataTransfer:
  6.     def __init__(self, target_address, target_name, port=1):
  7.         self.target_address = target_address
  8.         self.target_name = target_name
  9.         self.port = port
  10.         self.socket = None
  12.     def connect(self):
  13.         """
  14.         连接蓝牙设备
  15.         :return:
  16.         """
  17.         try:
  18.             self.socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
  19.             self.socket.connect((self.target_address, self.port))
  20.             print("Connected successfully. {} ({})".format(
  21.                 self.target_name, self.target_address
  22.             ))
  23.             return True
  24.         except bluetooth.BluetoothError as e:
  25.             self.socket = None
  26.             print("Connection failed:", str(e))
  28.         return False
  30.     def disconnect(self):
  31.         """
  32.         断开蓝牙连接
  33.         :return:
  34.         """
  35.         if self.socket is not None:
  36.             self.socket.close()
  37.             print("Disconnected.")
  39.     def send_data(self, data):
  40.         """
  41.         发送数据
  42.         :param data:
  43.         :return:
  44.         """
  45.         if self.socket is not None:
  46.             try:
  47.                 self.socket.send(data)
  48.                 print("Data sent:", data)
  49.                 return True
  50.             except bluetooth.BluetoothError as e:
  51.                 print("Failed to send data:", str(e))
  52.         else:
  53.             print("Not connected.")
  54.         return False
  56.     def receive_data(self, buffer_size=1024):
  57.         """
  58.         接收数据
  59.         :param buffer_size:
  60.         :return:
  61.         """
  62.         if self.socket is not None:
  63.             try:
  64.                 data = self.socket.recv(buffer_size)
  65.                 print("Data received:", data.decode())
  66.                 return data
  67.             except bluetooth.BluetoothError as e:
  68.                 print("Failed to receive data:", str(e))
  69.         else:
  70.             print("Not connected.")
  72.     @staticmethod
  73.     def scan_devices():
  74.         """
  75.         扫描所有蓝牙设备
  76.         :return:
  77.         """
  78.         devices = bluetooth.discover_devices()
  79.         print("Scanning devices...")
  80.         device_list = []
  81.         for addr in devices:
  82.             name = bluetooth.lookup_name(addr)
  83.             print("Found device:", name, "(", addr, ")")
  84.             device_list.append((addr, name))
  85.         return device_list
  88. if __name__ == '__main__':
  89.     devices = BluetoothDataTransfer.scan_devices()
  90.     for device in devices:
  91.         print(device)
  93.     # 示例用法
  94.     bd = BluetoothDataTransfer("48:E7:DA:AA:98:28", "LIN", 1)  # 替换为目标设备的蓝牙地址和端口号
  95.     if bd.connect():
  96.         bd.send_data("Hello, Bluetooth!")  # 发送数据
  97.         bd.receive_data()  # 接收数据
  98.         bd.disconnect()