Python生成OpenSignals数据时间戳写入InfluxDB遇TypeError问题排查

问题背景 #

有一个OpenSignals格式的文本文件，前3行为头部信息：

# OpenSignals Text File Format. Version 1
# {"EC:1B:BD:62:F6:A2": {"date": "2022-7-22", "time": "10:43:47.162", "label": ["A1", "A2", "ECG"]}}
# EndOfHeader
0   1   512
0   0   500
1   1   512
0   0   525
0   0   516
1   2   499
1   1   501
1   0   512
1   0   496
0   0   525
0   0   525
0   0   512

第二行包含测量起始日期和时间，第4行开始是对应A1、A2、ECG标签的数据，采样间隔1毫秒。需求是基于起始时间生成时间戳：第一行加0毫秒，第二行加2毫秒，第三行加3毫秒，以此类推，将时间戳作为索引写入InfluxDB时出现错误：

TypeError: Must be DataFrame with DatetimeIndex or PeriodIndex.

用户的代码如下：

import pandas as pd
import numpy as np
import linecache
import json
import time
from datetime import datetime,timedelta
from influxdb import DataFrameClient

bhost='192.168.1.150'
dbport=8086
dbuser='telegraf'
dbpasswd="pass"
dbname='test1'
protocol='line'

#SELECT JSON FROM SECOND LINE
info=(linecache.getline('bio.txt',2))[24:-2]
info_json=json.loads(info)
head_info=info_json["label"]

# load Data and add delta column 
df=pd.read_table('bio.txt',skiprows=3,header=None,names=head_info)

# Date, Time end Delta
df['date']=info_json["date"]
df['time']=info_json["time"]
df.insert(0,'delta',range(0,len(df)))
format='%Y-%m-%d %H:%M:%S.%f'
df['Datetime']=pd.to_datetime(df['date']+ ' '+df['time'], format=format)

# Create Index
df['timestamp']=pd.to_datetime(date_time_obj+(pd.to_timedelta(df.delta,unit='ms')))
df=df.set_index(pd.DatetimeIndex(df['timestamp']))
df = df.drop(['date','time','delta','Datetime'], axis=1)

#save to InfluxDb
client=DataFrameClient(dbhost,dbport,dbuser,dbpasswd,dbname)
client.write_points(dn,"signal",time_precision='ms')

错误原因分析 #

1. 未定义变量date_time_obj：代码中直接使用该变量但未赋值，导致生成timestamp时出错，最终索引无法成为有效的DatetimeIndex。
2. 变量名错误：写入InfluxDB时误用dn，实际应传递DataFrame对象df。
3. 数据读取分隔符问题：pd.read_table默认用制表符分隔，但源数据是空格分隔，可能导致列解析错误。
4. 时间戳生成逻辑冗余：每行重复计算起始时间，效率低且易引发格式问题。

修正后的代码 #

import pandas as pd
import linecache
import json
from influxdb import DataFrameClient

# InfluxDB配置
dbhost='192.168.1.150'
dbport=8086
dbuser='telegraf'
dbpasswd="pass"
dbname='test1'

# 读取头部信息
info_line = linecache.getline('bio.txt', 2)
info_json = json.loads(info_line.strip()[2:])
device_info = next(iter(info_json.values()))
head_info = device_info["label"]

# 加载数据：指定空格分隔符，确保列解析正确
df = pd.read_table('bio.txt', skiprows=3, header=None, names=head_info, sep='\s+')

# 生成符合需求的时间戳
start_time = pd.to_datetime(f"{device_info['date']} {device_info['time']}", format='%Y-%m-%d %H:%M:%S.%f')
# 按要求构造增量：第一行0ms，第二行2ms，第三行3ms，后续依次+1ms
delta_ms = [0] + [i+1 for i in range(1, len(df))]
df['timestamp'] = start_time + pd.to_timedelta(delta_ms, unit='ms')

# 设置DatetimeIndex
df.set_index('timestamp', inplace=True)

# 写入InfluxDB
client = DataFrameClient(dbhost, dbport, dbuser, dbpasswd, dbname)
client.write_points(df, "signal", time_precision='ms')

关键修正点说明 #

• 修复未定义变量问题：直接从设备信息提取起始时间，简化时间戳生成逻辑。
• 修正数据读取：添加sep='\s+'确保空格分隔的数据正确解析。
• 严格按照需求构造毫秒增量，避免时间戳逻辑错误。
• 修正变量名错误，传递正确的DataFrame对象给InfluxDB客户端。
• 简化代码流程，确保最终DataFrame的索引是有效的DatetimeIndex。

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