如何将Python人脸追踪脚本中动态变化的变量导入至另一个脚本？

这里有几个靠谱的方法，你可以根据自己的使用场景来选择：

方法1：封装成生成器函数（同步调用场景） #

如果你的另一个脚本需要在同一个进程里调用人脸追踪逻辑，同时实时获取变量，把人脸追踪代码封装成生成器函数是最简单的方式。生成器可以在循环中不断吐出最新的变量值，供消费脚本迭代获取。

修改人脸追踪脚本（命名为face_tracker.py）： #

把原来的循环逻辑封装成生成器，用yield返回实时计算的变量：

import cv2
import time
# 确保你已经定义了face_data, Distance_finder, averageFinder, speedFinder这些辅助函数，以及相关常量

def track_face_speed():
    cap = cv2.VideoCapture(0)  # 按你的实际视频源调整
    out = cv2.VideoWriter('output.avi', cv2.VideoWriter_fourcc(*'XVID'), 20.0, (640, 480))
    intialDisntace = 0
    changeInTime = 0
    DistanceList = []
    speedList = []
    Focal_length_found = 500  # 替换成你的实际焦距值
    Known_width = 15  # 替换成你的已知人脸宽度
    ORANGE = (0, 165, 255)
    GREEN = (0, 255, 0)
    BLACK = (0, 0, 0)
    fonts = cv2.FONT_HERSHEY_COMPLEX_SMALL

    while True:
        _, frame = cap.read()
        intialTime = time.time()
        # 初始化Distance_level为列表最后一个值（如果有），否则为0
        distance_level = int(DistanceList[-1]) if DistanceList else 0
        face_width_in_frame, Faces, FC_X, FC_Y = face_data(frame, True, distance_level)
        
        averageSpeed = 0
        roundedDistance = 0

        for (face_x, face_y, face_w, face_h) in Faces:
            if face_width_in_frame != 0:
                Distance = Distance_finder(Focal_length_found, Known_width, face_width_in_frame)
                DistanceList.append(Distance)
                avergDistnce = averageFinder(DistanceList, 6)
                roundedDistance = round((avergDistnce * 0.0254), 2)

                if intialDisntace != 0:
                    changeDistance = Distance - intialDisntace
                    distanceInMeters = changeDistance * 0.0254
                    velocity = speedFinder(distanceInMeters, changeInTime)
                    speedList.append(velocity)
                    averageSpeed = averageFinder(speedList, 6)
                    intialDisntace = avergDistnce
                    changeInTime = time.time() - intialTime
                else:
                    # 第一次检测到人脸时初始化距离和时间
                    intialDisntace = avergDistnce
                    changeInTime = time.time() - intialTime

        # 处理负速度的情况
        if averageSpeed < 0:
            averageSpeed = averageSpeed * -1
        
        # 吐出最新的变量值
        yield averageSpeed, roundedDistance

        # 原有的显示和保存逻辑
        cv2.line(frame, (25, 45), (80, 45), ORANGE, 26)
        cv2.line(frame, (25, 45), (80, 45), GREEN, 20)
        cv2.putText(frame, f"Speed: {round(averageSpeed,2)} in/s", (30, 50), fonts, 0.5, BLACK, 2)
        text_pos = (FC_X-6, FC_Y-6) if FC_X and FC_Y else (10, 10)
        cv2.putText(frame, f"Distance {roundedDistance*39.57} inches", text_pos, fonts, 0.5, BLACK, 2)
        cv2.imshow("frame", frame)
        out.write(frame)

        if cv2.waitKey(1) == ord("q"):
            break

    cap.release()
    out.release()
    cv2.destroyAllWindows()

消费脚本（命名为data_consumer.py）： #

导入生成器，循环获取实时变量：

from face_tracker import track_face_speed

# 迭代生成器，实时获取数据
for speed, distance in track_face_speed():
    print(f"当前速度: {round(speed,2)} in/s, 当前距离: {distance} 米")
    # 在这里添加你需要用这些变量做的业务逻辑

方法2：使用多进程队列（独立进程实时传输） #

如果你的人脸追踪脚本需要单独运行，同时另一个脚本也要实时获取数据，用multiprocessing.Queue是最优解——它是进程安全的，能高效地在两个独立进程间传递数据。

修改人脸追踪脚本（face_tracker.py）： #

把追踪逻辑封装成进程工作函数，将变量放入队列：

import cv2
import time
from multiprocessing import Process, Queue

# 辅助函数和常量省略，和方法1一致

def face_tracker_worker(queue):
    cap = cv2.VideoCapture(0)
    out = cv2.VideoWriter('output.avi', cv2.VideoWriter_fourcc(*'XVID'), 20.0, (640, 480))
    intialDisntace = 0
    changeInTime = 0
    DistanceList = []
    speedList = []
    Focal_length_found = 500
    Known_width = 15
    ORANGE = (0, 165, 255)
    GREEN = (0, 255, 0)
    BLACK = (0, 0, 0)
    fonts = cv2.FONT_HERSHEY_COMPLEX_SMALL

    while True:
        _, frame = cap.read()
        intialTime = time.time()
        distance_level = int(DistanceList[-1]) if DistanceList else 0
        face_width_in_frame, Faces, FC_X, FC_Y = face_data(frame, True, distance_level)
        
        averageSpeed = 0
        roundedDistance = 0

        for (face_x, face_y, face_w, face_h) in Faces:
            if face_width_in_frame != 0:
                Distance = Distance_finder(Focal_length_found, Known_width, face_width_in_frame)
                DistanceList.append(Distance)
                avergDistnce = averageFinder(DistanceList, 6)
                roundedDistance = round((avergDistnce * 0.0254), 2)

                if intialDisntace != 0:
                    changeDistance = Distance - intialDisntace
                    distanceInMeters = changeDistance * 0.0254
                    velocity = speedFinder(distanceInMeters, changeInTime)
                    speedList.append(velocity)
                    averageSpeed = averageFinder(speedList, 6)
                    intialDisntace = avergDistnce
                    changeInTime = time.time() - intialTime
                else:
                    intialDisntace = avergDistnce
                    changeInTime = time.time() - intialTime

        if averageSpeed < 0:
            averageSpeed = averageSpeed * -1
        
        # 将数据放入队列，避免队列满导致阻塞
        try:
            queue.put_nowait((averageSpeed, roundedDistance))
        except queue.Full:
            # 队列满时丢弃旧数据，放入新数据
            queue.get()
            queue.put((averageSpeed, roundedDistance))

        # 显示逻辑省略，和方法1一致
        cv2.line(frame, (25, 45), (80, 45), ORANGE, 26)
        cv2.line(frame, (25, 45), (80, 45), GREEN, 20)
        cv2.putText(frame, f"Speed: {round(averageSpeed,2)} in/s", (30, 50), fonts, 0.5, BLACK, 2)
        text_pos = (FC_X-6, FC_Y-6) if FC_X and FC_Y else (10, 10)
        cv2.putText(frame, f"Distance {roundedDistance*39.57} inches", text_pos, fonts, 0.5, BLACK, 2)
        cv2.imshow("frame", frame)
        out.write(frame)

        if cv2.waitKey(1) == ord("q"):
            break

    cap.release()
    out.release()
    cv2.destroyAllWindows()
    # 发送结束信号
    queue.put(None)

消费脚本（data_consumer.py）： #

启动追踪进程，同时从队列中获取数据：

from multiprocessing import Queue, Process
from face_tracker import face_tracker_worker

def consume_data(queue):
    while True:
        data = queue.get()
        if data is None:
            # 接收到结束信号，退出循环
            break
        averageSpeed, roundedDistance = data
        print(f"当前速度: {round(averageSpeed,2)} in/s, 当前距离: {roundedDistance} 米")
        # 添加你的业务逻辑

if __name__ == "__main__":
    # 设置队列最大容量，避免内存占用过高
    queue = Queue(maxsize=10)
    # 启动人脸追踪进程
    tracker_process = Process(target=face_tracker_worker, args=(queue,))
    tracker_process.start()
    # 启动数据消费进程
    consumer_process = Process(target=consume_data, args=(queue,))
    consumer_process.start()
    # 等待进程结束
    tracker_process.join()
    consumer_process.join()

方法3：使用本地文件（简单场景，实时性一般） #

如果你的场景对实时性要求不高，或者不想搞复杂的进程通信，可以把变量写入一个JSON临时文件，另一个脚本定期读取这个文件。这种方法实现最简单，但可能有轻微延迟。

修改人脸追踪脚本的循环部分： #

在计算完变量后，写入JSON文件：

# 在循环内计算完averageSpeed和roundedDistance后
import json
with open("face_data.json", "w") as f:
    json.dump({
        "averageSpeed": averageSpeed,
        "roundedDistance": roundedDistance
    }, f)

消费脚本： #

定期读取文件获取数据：

import json
import time

while True:
    try:
        with open("face_data.json", "r") as f:
            data = json.load(f)
            averageSpeed = data["averageSpeed"]
            roundedDistance = data["roundedDistance"]
            print(f"当前速度: {round(averageSpeed,2)} in/s, 当前距离: {roundedDistance} 米")
    except FileNotFoundError:
        print("等待人脸追踪脚本生成数据...")
    # 控制读取频率，避免频繁IO操作
    time.sleep(0.1)

方法4：使用本地Socket（跨进程/跨机器场景） #

如果需要跨机器传输数据，或者更灵活的通信方式，可以用本地TCP Socket。人脸追踪脚本作为服务器发送数据，消费脚本作为客户端接收。

人脸追踪服务器脚本（face_tracker.py）： #

import cv2
import time
import socket
import json

# 辅助函数和常量省略

def main():
    # 创建本地TCP服务器
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 12345))
    server_socket.listen(1)
    print("等待客户端连接...")
    conn, addr = server_socket.accept()
    print(f"客户端已连接: {addr}")

    cap = cv2.VideoCapture(0)
    out = cv2.VideoWriter('output.avi', cv2.VideoWriter_fourcc(*'XVID'), 20.0, (640, 480))
    intialDisntace = 0
    changeInTime = 0
    DistanceList = []
    speedList = []
    Focal_length_found = 500
    Known_width = 15
    ORANGE = (0, 165, 255)
    GREEN = (0, 255, 0)
    BLACK = (0, 0, 0)
    fonts = cv2.FONT_HERSHEY_COMPLEX_SMALL

    try:
        while True:
            _, frame = cap.read()
            intialTime = time.time()
            distance_level = int(DistanceList[-1]) if DistanceList else 0
            face_width_in_frame, Faces, FC_X, FC_Y = face_data(frame, True, distance_level)
            
            averageSpeed = 0
            roundedDistance = 0

            for (face_x, face_y, face_w, face_h) in Faces:
                if face_width_in_frame != 0:
                    Distance = Distance_finder(Focal_length_found, Known_width, face_width_in_frame)
                    DistanceList.append(Distance)
                    avergDistnce = averageFinder(DistanceList, 6)
                    roundedDistance = round((avergDistnce * 0.0254), 2)

                    if intialDisntace != 0:
                        changeDistance = Distance - intialDisntace
                        distanceInMeters = changeDistance * 0.0254
                        velocity = speedFinder(distanceInMeters, changeInTime)
                        speedList.append(velocity)
                        averageSpeed = averageFinder(speedList, 6)
                        intialDisntace = avergDistnce
                        changeInTime = time.time() - intialTime
                    else:
                        intialDisntace = avergDistnce
                        changeInTime = time.time() - intialTime

            if averageSpeed < 0:
                averageSpeed = averageSpeed * -1
            
            # 发送数据，添加换行符作为消息分隔符
            data = json.dumps({
                "averageSpeed": averageSpeed,
                "roundedDistance": roundedDistance
            }) + "\n"
            conn.sendall(data.encode('utf-8'))

            # 显示逻辑省略
            cv2.line(frame, (25, 45), (80, 45), ORANGE, 26)
            cv2.line(frame, (25, 45), (80, 45), GREEN, 20)
            cv2.putText(frame, f"Speed: {round(averageSpeed,2)} in/s", (30, 50), fonts, 0.5, BLACK, 2)
            text_pos = (FC_X-6, FC_Y-6) if FC_X and FC_Y else (10, 10)
            cv2.putText(frame, f"Distance {roundedDistance*39.57} inches", text_pos, fonts, 0.5, BLACK, 2)
            cv2.imshow("frame", frame)
            out.write(frame)

            if cv2.waitKey(1) == ord("q"):
                break
    finally:
        conn.close()
        server_socket.close()
        cap.release()
        out.release()
        cv2.destroyAllWindows()

if __name__ == "__main__":
    main()

消费客户端脚本： #

import socket
import json

def main():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(('localhost', 12345))
    buffer = ""

    try:
        while True:
            data = client_socket.recv(1024).decode('utf-8')
            if not data:
                break
            buffer += data
            # 按换行符分割完整的消息
            while "\n" in buffer:
                message, buffer = buffer.split("\n", 1)
                try:
                    data_dict = json.loads(message)
                    averageSpeed = data_dict["averageSpeed"]
                    roundedDistance = data_dict["roundedDistance"]
                    print(f"当前速度: {round(averageSpeed,2)} in/s, 当前距离: {roundedDistance} 米")
                except json.JSONDecodeError:
                    print("接收到无效数据")
    finally:
        client_socket.close()

if __name__ == "__main__":
    main()

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