如何通过wfdb.wrsamp获取字节数据而非写入磁盘以上传至S3？

可以通过直接调用wfdb的底层写入函数，将内容写入内存缓冲区而非磁盘，完全避免磁盘IO操作，之后直接读取缓冲区字节内容上传至S3。

核心思路 #

wfdb的上层wrsamp函数硬编码了磁盘写入逻辑，但它依赖的底层write_header（写入头文件）和write_signal（写入信号数据）函数支持传入文件对象，而非仅文件路径。我们可以利用这一点，用io.BytesIO创建内存缓冲区，直接将内容写入其中。

具体实现代码 #

import io
import wfdb
import boto3
from wfdb.io._header import write_header
from wfdb.io._signal import write_signal

# 1. 准备你的医疗信号数据和头信息（示例）
sig = ...  # 你的信号数据数组，格式需符合wfdb要求
sampling_freq = 250  # 采样频率
signal_names = ["ECG_Lead1", "ECG_Lead2"]
signal_units = ["mV", "mV"]
record_comments = ["Processed for S3 upload"]

# 2. 创建内存缓冲区存储头文件和信号文件
hea_buffer = io.BytesIO()
dat_buffer = io.BytesIO()

# 3. 写入头信息到内存缓冲区
write_header(
    file=hea_buffer,
    fs=sampling_freq,
    sig_len=len(sig),
    n_sig=len(signal_names),
    sig_name=signal_names,
    units=signal_units,
    comments=record_comments
)

# 4. 写入信号数据到内存缓冲区
# fmt参数根据你的数据格式选择，比如"16"代表16位整数，"f"代表浮点等
write_signal(
    file=dat_buffer,
    sig=sig,
    fmt="16",
    byteorder="little"  # 字节序根据需求调整
)

# 5. 获取缓冲区的字节内容
hea_bytes = hea_buffer.getvalue()
dat_bytes = dat_buffer.getvalue()

# 6. 直接上传至S3
s3_client = boto3.client("s3")
s3_client.put_object(Bucket="your-bucket-name", Key="record.hea", Body=hea_bytes)
s3_client.put_object(Bucket="your-bucket-name", Key="record.dat", Body=dat_bytes)

备选方案：猴子补丁适配上层wrsamp接口 #

如果必须使用wfdb.wrsamp的上层接口，可以通过临时替换wfdb内部的文件打开函数，让其写入内存缓冲区。这种方法依赖wfdb内部实现，版本更新可能失效，仅作备选：

import io
import wfdb
import boto3
from contextlib import contextmanager

# 定义内存版的open函数，将内容写入BytesIO
@contextmanager
def memory_file_handler(filepath, mode):
    buffer = io.BytesIO()
    yield buffer
    # 将缓冲区内容存入全局变量，方便后续获取
    globals()[filepath.split("/")[-1]] = buffer.getvalue()

# 临时替换wfdb底层模块的open函数
original_header_open = wfdb.io._header.open
original_signal_open = wfdb.io._signal.open
wfdb.io._header.open = memory_file_handler
wfdb.io._signal.open = memory_file_handler

try:
    # 调用wrsamp，传入任意record_name，实际写入内存
    wfdb.wrsamp(
        record_name="temp_ecg",
        fs=250,
        sig=sig,
        sig_name=["ECG_Lead1", "ECG_Lead2"],
        units=["mV", "mV"]
    )
    # 获取内存中的字节内容
    hea_bytes = globals()["temp_ecg.hea"]
    dat_bytes = globals()["temp_ecg.dat"]
    # 上传至S3
    s3_client = boto3.client("s3")
    s3_client.put_object(Bucket="your-bucket-name", Key="temp_ecg.hea", Body=hea_bytes)
    s3_client.put_object(Bucket="your-bucket-name", Key="temp_ecg.dat", Body=dat_bytes)
finally:
    # 恢复原open函数，避免影响后续代码
    wfdb.io._header.open = original_header_open
    wfdb.io._signal.open = original_signal_open

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