测量系统中基于Python实现串口多路复用/解复用及虚拟串口构建问询

Great question—this is exactly the kind of problem that can be solved cleanly with Python without needing extra hardware. Let's break down how to implement virtual serial ports for your three instruments, so your existing code works unchanged while respecting the A/B mutual exclusion rule.

First, let's recap your setup to make sure I'm aligned: You've got instruments A, B, C connected through a PIC24 that adds A_/B_/C_ prefixes to all commands/responses, funneling everything to a single physical serial port on your BBB. Your A and B share a Python module with a measure() function that expects a serial port object, and you need to let different parts of your code interact with each instrument as if they were separate ports.

Core Approach #

We'll build two key components:

1. A Shared Serial Manager that handles the physical serial port, enforces the A/B mutual exclusion rule, and routes incoming data to the correct instrument's queue.
2. Virtual Serial Wrappers for each instrument that mimic the serial.Serial interface (so your existing measure() function works without modifications). These wrappers automatically add the right prefix to outgoing commands and filter responses to only return data for their instrument.

Step 1: Shared Serial Manager #

This class manages the physical serial connection, listens for incoming data, and routes it to the appropriate response queue. It also uses a mutex to prevent A and B from running concurrently.

import serial
import threading
from queue import Queue

class SharedSerialManager:
    def __init__(self, port, baudrate=9600, timeout=1):
        self.physical_serial = serial.Serial(port, baudrate=baudrate, timeout=timeout)
        self.a_b_mutex = threading.Lock()  # Enforce A/B mutual exclusion
        self.response_queues = {'A': Queue(), 'B': Queue(), 'C': Queue()}
        
        # Start background thread to listen for incoming data
        self.listener_thread = threading.Thread(target=self._route_incoming_data, daemon=True)
        self.listener_thread.start()

    def _route_incoming_data(self):
        """Listen for serial input, split prefix and data, send to correct queue."""
        while True:
            line = self.physical_serial.readline().decode('utf-8').strip()
            if not line or '_' not in line:
                continue  # Skip empty lines or malformed data
            
            prefix, payload = line.split('_', 1)
            if prefix in self.response_queues:
                self.response_queues[prefix].put(payload)

    def send_command(self, prefix, command):
        """Send a command with the instrument's prefix."""
        full_cmd = f"{prefix}_{command}\r\n"
        self.physical_serial.write(full_cmd.encode('utf-8'))

    def get_response(self, prefix):
        """Get the next response for the specified instrument (blocks until timeout)."""
        try:
            return self.response_queues[prefix].get(timeout=self.physical_serial.timeout)
        except:
            return None

    # Helper methods for A/B mutex control
    def lock_a_b(self):
        self.a_b_mutex.acquire()
    
    def unlock_a_b(self):
        self.a_b_mutex.release()

Step 2: Virtual Serial Port Wrapper #

This class mimics the serial.Serial interface so your existing measure() function can use it as if it were a real serial port. It handles prefixing commands, filtering responses, and automatically managing the A/B mutex via a context manager.

class VirtualSerialPort:
    def __init__(self, manager, instrument_prefix):
        self.manager = manager
        self.prefix = instrument_prefix
        self.is_a_b = instrument_prefix in ('A', 'B')

    def write(self, command):
        """Mimic serial.write(): send command with prefix."""
        # Convert bytes to string if needed
        cmd_str = command.decode('utf-8').strip() if isinstance(command, bytes) else str(command).strip()
        
        # Ensure A/B ports are locked before sending
        if self.is_a_b and not self.manager.a_b_mutex.locked():
            raise RuntimeError("A/B virtual port must be used within a context manager (with statement)")
        
        self.manager.send_command(self.prefix, cmd_str)

    def readline(self):
        """Mimic serial.readline(): return filtered response as bytes."""
        response = self.manager.get_response(self.prefix)
        return f"{response}\r\n".encode('utf-8') if response else b''

    # Add other serial methods your code uses (example below)
    def read(self, size=1):
        """Simplified read method; adjust based on your needs."""
        return self.readline()

    def flush(self):
        """No-op for virtual port (physical port is managed by the manager)."""
        pass

    # Context manager to handle A/B mutex automatically
    def __enter__(self):
        if self.is_a_b:
            self.manager.lock_a_b()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.is_a_b:
            self.manager.unlock_a_b()

Step 3: Integrate with Your Existing Code #

Now you can create virtual ports for each instrument and pass them to your measure() function just like real serial ports.

# Initialize the shared manager with your physical port (e.g., /dev/ttyO4)
serial_manager = SharedSerialManager('/dev/ttyO4', baudrate=9600)

# Create virtual ports for each instrument
virtual_a = VirtualSerialPort(serial_manager, 'A')
virtual_b = VirtualSerialPort(serial_manager, 'B')
virtual_c = VirtualSerialPort(serial_manager, 'C')

# Example 1: Use instrument A with automatic mutex handling
with virtual_a as port_a:
    result_a = measure(port_a)  # Your existing function works unchanged!

# Example 2: Use instrument C (no mutex needed)
result_c = measure(virtual_c)

# Example 3: Run A and C in parallel (allowed per your requirements)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
    future_a = executor.submit(lambda: measure(virtual_a))
    future_c = executor.submit(lambda: measure(virtual_c))
    result_a_parallel = future_a.result()
    result_c_parallel = future_c.result()

Key Notes #

• Concurrency Safety: The context manager for A/B ports automatically handles locking/unlocking, so you never have to worry about accidental concurrent use.
• Timing: Since you're okay with hundreds of milliseconds of delay, the queue-based approach with timeouts works perfectly—no latency issues here.
• Extensibility: If your measure() function uses other serial.Serial methods (like in_waiting), just add them to the VirtualSerialPort class to mimic the behavior.
• Testing: You can test this setup with your existing multi-port test rig by creating a mock SharedSerialManager that routes data without the PIC24.

This approach keeps all logic in Python, avoids extra hardware, and plays nicely with your existing codebase.

内容的提问来源于stack exchange，提问作者Dieter Vansteenwegen