求助：用Python同步运行数据采集程序并配置VitalSignsCapture参数

一、配置VitalSignsCapture应用内选项 #

假设VitalSignsCapture是可实例化的类，可通过初始化参数直接传入6项配置。若该类提供独立配置方法（如set_config()），也可改用对应方式传入：

# 定义ICU监护仪的6项配置（替换为你的实际参数）
icu_config = {
    "eth_interface": "eth0",
    "data_sample_rate": 500,
    "monitored_signals": ["heart_rate", "bp_sys", "bp_dia", "spo2", "temperature", "respiration"],
    "connection_timeout": 10,
    "buffer_size": 8192,
    "noise_filter": True
}

# 方式1：通过初始化参数传入配置
capture_device = VitalSignsCapture(**icu_config)

# 方式2：若类提供set_config方法
# capture_device = VitalSignsCapture()
# capture_device.set_config(icu_config)

二、异步启动双采集任务 #

使用concurrent.futures.ThreadPoolExecutor实现两个任务并行执行，确保时间戳同步。核心是同时触发两类数据采集，统一记录时间戳。

1. 编写采集函数 #

import csv
import time
import requests
from concurrent.futures import ThreadPoolExecutor

def capture_icu_data(capture_device, timestamp):
    # 使用统一传入的时间戳，保证与HTTP采集同步
    icu_raw_data = capture_device.fetch_latest_data()  # 假设VitalSignsCapture的采集方法
    return {"timestamp": timestamp, **icu_raw_data}

def capture_http_data(http_url, timestamp):
    # 使用同一时间戳
    response = requests.get(http_url, timeout=5)
    response.raise_for_status()
    http_raw_data = response.json()
    return {"timestamp": timestamp, **http_raw_data}

2. 并行执行并保存CSV #

def save_combined_data(icu_result, http_result, csv_path):
    # 合并两类数据，自动对齐字段
    combined_data = {**icu_result, **http_result}
    fieldnames = list(combined_data.keys())

    # 写入CSV，自动创建表头（首次写入时）
    with open(csv_path, mode='a', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if f.tell() == 0:
            writer.writeheader()
        writer.writerow(combined_data)

def main():
    # 初始化配置与参数
    icu_config = {
        "eth_interface": "eth0",
        "data_sample_rate": 500,
        "monitored_signals": ["heart_rate", "bp_sys", "bp_dia", "spo2", "temperature", "respiration"],
        "connection_timeout": 10,
        "buffer_size": 8192,
        "noise_filter": True
    }
    capture_device = VitalSignsCapture(**icu_config)
    http_endpoint = "http://your-data-source/api/real-time-data"
    csv_output = "icu_http_combined_data.csv"

    # 生成统一时间戳，确保两类数据时间完全对齐
    capture_timestamp = time.time()

    # 并行启动两个采集任务
    with ThreadPoolExecutor(max_workers=2) as executor:
        icu_task = executor.submit(capture_icu_data, capture_device, capture_timestamp)
        http_task = executor.submit(capture_http_data, http_endpoint, capture_timestamp)

        # 获取任务结果
        icu_data = icu_task.result()
        http_data = http_task.result()

    # 保存合并后的数据
    save_combined_data(icu_data, http_data, csv_output)
    print(f"采集完成，数据已保存至 {csv_output}")

if __name__ == "__main__":
    main()

三、关键优化点 #

• 时间戳精度：提前生成统一时间戳并传入两个任务，避免线程调度导致的时间差，确保数据时间完全同步。
• 异常处理：可在采集函数中添加try-except块，处理ICU设备连接失败、HTTP请求超时等异常，例如：

  def capture_http_data(http_url, timestamp):
      try:
          response = requests.get(http_url, timeout=5)
          response.raise_for_status()
          return {"timestamp": timestamp, **response.json()}
      except requests.exceptions.RequestException as e:
          return {"timestamp": timestamp, "http_error": str(e)}
  

• 持续采集：若需要长期持续采集，可在main中添加循环，配合time.sleep()控制采集间隔。

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