多并行Python程序通信方案求助：CNC自动化测试场景

看起来你遇到了典型的跨进程状态共享问题——每次启动新脚本都重新初始化Cnc对象，自然会丢设备位置这类关键状态。下面给你几个实用的解决方案，从简单到复杂都有，你可以根据自己的场景选择：

方案1：把CNC做成常驻HTTP服务（最推荐） #

这是工业控制场景里最常用的思路：把控制CNC的逻辑封装成一个长期运行的后台服务，其他程序（不管是Python脚本还是C# UI）都通过调用服务接口来发指令、查状态，全程只维护一个Cnc实例。

• 具体实现步骤：

  1. 用Python的FastAPI或Flask写一个轻量HTTP服务，在树莓派上常驻运行。服务启动时初始化唯一的Cnc实例，把设备状态存在内存里。
  2. 定义几个核心API端点：比如/get-position返回当前坐标、/move接收目标位置参数执行移动、/start-test触发自动化测试流程。
  3. 你的C# UI只需要用HttpClient发HTTP请求调用这些接口就行，完全不用管Cnc对象的初始化问题。

• 优点：

  • 状态全程持久化，服务不重启就不会丢数据；
  • 跨语言友好，C#调用HTTP接口毫无门槛；
  • 扩展性强，后续加新功能只需要加新的API端点。

• 示例代码片段：

from fastapi import FastAPI
from your_cnc_module import Cnc

# 只初始化一次，常驻内存
cnc_instance = Cnc()
app = FastAPI()

# 获取当前位置接口
@app.get("/get-position")
def get_position():
    return {
        "x": cnc_instance.x_pos,
        "y": cnc_instance.y_pos,
        "z": cnc_instance.z_pos
    }

# 移动到指定位置接口
@app.post("/move")
def move_to(x: float, y: float, z: float):
    cnc_instance.move_to(x, y, z)
    return {
        "status": "success",
        "new_position": {
            "x": cnc_instance.x_pos,
            "y": cnc_instance.y_pos,
            "z": cnc_instance.z_pos
        }
    }

启动服务用uvicorn main:app --host 0.0.0.0，让树莓派在局域网内可访问，C#里直接发GET/POST请求就能控制CNC了。

方案2：本地进程间通信（IPC） #

如果不想用HTTP，也可以用Python原生的IPC机制，比如本地套接字、管道，适合树莓派内部多个Python进程之间的通信。

• 具体实现步骤：

  1. 启动一个主进程，初始化Cnc实例，然后监听本地套接字；
  2. 其他子进程或外部程序通过套接字发送JSON格式的指令，主进程接收后调用对应的Cnc方法，再把结果返回。

• 优点：比HTTP更轻量，延迟更低，适合对响应速度要求高的场景。

• 示例代码（本地套接字）：
  主进程（常驻服务）：

import socket
import json
from your_cnc_module import Cnc

cnc = Cnc()
# 创建本地套接字
server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server_socket.bind("/tmp/cnc_socket")
server_socket.listen(1)

while True:
    conn, _ = server_socket.accept()
    data = conn.recv(1024).decode()
    cmd = json.loads(data)
    
    if cmd["action"] == "move":
        cnc.move_to(cmd["x"], cmd["y"], cmd["z"])
        result = {"status": "ok", "position": {"x": cnc.x_pos, "y": cnc.y_pos, "z": cnc.z_pos}}
    elif cmd["action"] == "get_pos":
        result = {"position": {"x": cnc.x_pos, "y": cnc.y_pos, "z": cnc.z_pos}}
    
    conn.send(json.dumps(result).encode())
    conn.close()

客户端脚本：

import socket
import json

client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client_socket.connect("/tmp/cnc_socket")

# 发送移动指令
cmd = json.dumps({"action": "move", "x": 10.0, "y": 20.0, "z": 5.0})
client_socket.send(cmd.encode())

# 接收结果
response = json.loads(client_socket.recv(1024).decode())
print(response)
client_socket.close()

方案3：状态文件持久化+锁机制 #

如果你的场景非常简单，不想跑常驻服务，可以把Cnc的状态（比如当前位置）保存到本地JSON/CSV文件，每次脚本启动时先读文件恢复状态，操作完成后再更新文件，同时用文件锁避免多进程冲突。

• 具体实现步骤：

  1. 初始化Cnc时，先读取状态文件加载历史位置；
  2. 执行移动操作后，把新位置写入文件；
  3. 用fcntl（Linux）或msvcrt（Windows）加文件锁，防止多个进程同时修改状态文件。

• 优点：实现简单，不需要额外服务；

• 缺点：有数据一致性风险（比如进程崩溃可能导致状态没更新），适合操作频率不高的场景。

• 示例代码：

import json
import fcntl

STATE_FILE = "/tmp/cnc_state.json"

class Cnc:
    def __init__(self):
        # 读取状态文件恢复位置
        try:
            with open(STATE_FILE, "r") as f:
                fcntl.flock(f, fcntl.LOCK_SH)  # 加共享锁读
                state = json.load(f)
                self.x_pos = state["x"]
                self.y_pos = state["y"]
                self.z_pos = state["z"]
                fcntl.flock(f, fcntl.LOCK_UN)
        except FileNotFoundError:
            # 默认初始位置
            self.x_pos = 0.0
            self.y_pos = 0.0
            self.z_pos = 0.0

    def move_to(self, x, y, z):
        # 执行实际移动逻辑...
        self.x_pos = x
        self.y_pos = y
        self.z_pos = z
        
        # 更新状态文件
        with open(STATE_FILE, "w") as f:
            fcntl.flock(f, fcntl.LOCK_EX)  # 加排他锁写
            json.dump({
                "x": self.x_pos,
                "y": self.y_pos,
                "z": self.z_pos
            }, f)
            fcntl.flock(f, fcntl.LOCK_UN)

方案选择建议 #

• 如果需要C# UI远程（局域网）控制，优先选方案1（HTTP服务），跨语言兼容性最好，维护成本最低；
• 如果只是树莓派内部多个Python进程通信，**方案2（本地IPC）**更轻量高效；
• 如果场景简单、操作频率低，**方案3（文件持久化）**可以快速落地，但要注意数据一致性问题。

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