如何实现两个无限循环脚本间的通信？（串口数据采集场景）

你可以通过以下几种简单实用的方式实现脚本间的电压数据传递，适配不同的使用场景：

原数据采集脚本（供参考） #

import serial
import keyboard
import time
import sys
import struct

# PSerial 3.5 is in use for this example
# Press key 'x' to EXIT

CONST_SER_PORT = 'COM11'   # get the com port from device manger and enter it here

serDataq = serial.Serial(
    port=CONST_SER_PORT,
    timeout=0.5
)

serDataq.write(b"stop\r")        # stop in case device was left scanning
serDataq.write(b"encode 0\r")    # set up the device for binary mode
serDataq.write(b"slist 0 0\r")   # scan list position 0 channel 0 thru channel 7
serDataq.write(b"slist 1 1\r")   # scan list position 0 channel 0 thru channel 7
# serDataq.write(b"slist 2 2\r")   # scan list position 0 channel 0 thru channel 7
serDataq.write(b"srate 6000\r") 
serDataq.write(b"dec 250\r") 
serDataq.write(b"deca 1\r")
serDataq.write(b"ps 0\r")        # if you modify sample rate, you need modify this accordingly

time.sleep(0.5)
while True:
    try:
        i = serDataq.in_waiting
        if i > 0:
            response = serDataq.read(i)
            break
    except:
        pass
 
serDataq.reset_input_buffer()
serDataq.write(b"start\r")       # start scanning

numofchannel = 1 # if you modify the slist, you need modify this accordingly
numofbyteperscan = 2 * numofchannel

while True:
    try:
        if keyboard.is_pressed('x'):    # if key 'x' is pressed, stop the scanning and terminate the program
            serDataq.write(b"stop\r")
            time.sleep(1)           
            serDataq.close()
            print("Good-Bye")
            break
        if keyboard.is_pressed('p'):    # if key 'p' is pressed, pause the scanning
            serDataq.write(b"stop\r")
            time.sleep(1)           
            print("Paused")
            
        if keyboard.is_pressed('s'):    # if key 's' is pressed, start the scanning
            serDataq.reset_input_buffer()
            serDataq.write(b"start\r")
            time.sleep(1)           
            print("Start")
            
        else:
            i = serDataq.in_waiting
            if (i // numofbyteperscan) > 0:
                # we always read in scans
                response = serDataq.read(i - i % numofbyteperscan)
                
                count = (i - i % numofbyteperscan) // 2
                
                response2 = bytearray(response)
                
                Channel = struct.unpack("<" + "h"*count, response2)

                # print all
                voltage = round((Channel[0]/32768*10), 3)
                print(voltage, end='\r')
                # ------------ 此处添加通信逻辑 ------------
    except:
        pass

方法1：共享本地文件（最简单，适合低频率读取） #

无需额外依赖，通过读写同一本地文本文件实现数据传递。

修改数据采集脚本 #

在voltage = round(...)行后添加写入逻辑：

# 将电压值写入文件，覆盖旧内容
with open("voltage_data.txt", "w") as f:
    f.write(str(voltage))

数据读取脚本 #

import time

def get_current_voltage():
    try:
        with open("voltage_data.txt", "r") as f:
            return float(f.read().strip())
    except (FileNotFoundError, ValueError):
        return None  # 处理文件不存在或格式错误的情况

# 示例：每隔2秒读取一次电压值
while True:
    volt = get_current_voltage()
    if volt is not None:
        print(f"当前电压值：{volt} V")
    time.sleep(2)

方法2：本地Socket通信（实时性高，适合高频交互） #

通过本地TCP Socket建立点对点通信，采集脚本作为服务器，读取脚本作为客户端，按需请求电压值。

修改数据采集脚本 #

顶部导入socket库，主循环外初始化Socket服务器，主循环内处理请求：

import socket

# 初始化本地Socket服务器
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 8888))  # 本地端口8888
server_socket.listen(1)
print("等待读取脚本连接...")
conn, addr = server_socket.accept()
print(f"已连接：{addr}")

# 主循环内，得到voltage后添加：
# 监听客户端请求并返回电压值
try:
    if conn.recv(1024).decode() == "get_volt":
        conn.send(str(voltage).encode())
except BrokenPipeError:
    # 客户端断开后重新等待连接
    conn.close()
    conn, addr = server_socket.accept()
    print(f"重新连接：{addr}")

数据读取脚本 #

import socket

def request_voltage():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        client_socket.connect(('localhost', 8888))
        client_socket.send(b"get_volt")
        volt_str = client_socket.recv(1024).decode()
        return float(volt_str)
    except (ConnectionRefusedError, ValueError):
        return None
    finally:
        client_socket.close()

# 示例：按需读取电压值
while True:
    # 替换为你的触发逻辑（如用户输入、定时任务）
    input("按回车键获取电压值...")
    volt = request_voltage()
    if volt is not None:
        print(f"获取到电压值：{volt} V")

方法3：共享内存（最快，适合高性能场景） #

利用Python内置的共享内存实现极低延迟的数据传递，适合对响应速度要求高的场景。

修改数据采集脚本 #

from multiprocessing import Value, Process
import time

# 定义共享变量（'d'表示双精度浮点数）
shared_voltage = Value('d', 0.0)

def data_collection(shared_volt):
    # 原数据采集的所有代码放在此处，最后修改电压赋值逻辑：
    voltage = round((Channel[0]/32768*10), 3)
    shared_volt.value = voltage
    print(voltage, end='\r')

if __name__ == "__main__":
    # 启动数据采集进程
    collect_process = Process(target=data_collection, args=(shared_voltage,))
    collect_process.start()
    
    # 保持主进程运行
    try:
        while collect_process.is_alive():
            time.sleep(1)
    except KeyboardInterrupt:
        collect_process.terminate()
        collect_process.join()

数据读取脚本 #

from multiprocessing import Value
from multiprocessing.shared_memory import SharedMemory
import time

def read_shared_voltage():
    try:
        # 连接到共享内存（名称需与采集脚本一致）
        shm = SharedMemory(name='shared_voltage')
        shared_volt = Value('d', lock=False, shared_memory=shm)
        return shared_volt.value
    except FileNotFoundError:
        return None

# 示例读取
while True:
    volt = read_shared_voltage()
    if volt is not None:
        print(f"共享内存中的电压值：{volt} V")
    time.sleep(1)

内容的提问来源于stack exchange，提问作者Isaac Zaiek