139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
分析官方APP的CSV文件，提取真实的脉冲时长
|
||
"""
|
||
|
||
import csv
|
||
import statistics
|
||
|
||
def parse_csv(filename):
|
||
"""解析CSV文件，提取脉冲时长"""
|
||
times = []
|
||
states = []
|
||
|
||
with open(filename, 'r', encoding='utf-8') as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
time_str = row['Time'].replace(' ms', '')
|
||
time_ms = float(time_str)
|
||
times.append(time_ms)
|
||
state = int(row['Ch01'])
|
||
states.append(state)
|
||
|
||
return times, states
|
||
|
||
def calculate_pulses(times, states):
|
||
"""计算脉冲时长（微秒）"""
|
||
pulses = []
|
||
|
||
# 从第二个时间戳开始计算差值
|
||
for i in range(1, len(times)):
|
||
duration_ms = times[i] - times[i-1]
|
||
duration_us = duration_ms * 1000 # 转换为微秒
|
||
pulses.append(int(duration_us))
|
||
|
||
return pulses
|
||
|
||
def analyze_pulses(pulses):
|
||
"""分析脉冲时长，找出引导码、数据码等"""
|
||
print(f"总脉冲数: {len(pulses)}")
|
||
print(f"\n前30个脉冲:")
|
||
for i, pulse in enumerate(pulses[:30]):
|
||
print(f" [{i}]: {pulse}μs")
|
||
|
||
# 过滤掉异常大的脉冲（可能是信号间隔）
|
||
normal_pulses = [p for p in pulses if p < 10000]
|
||
|
||
# 统计脉冲时长分布
|
||
pulse_counts = {}
|
||
for pulse in normal_pulses:
|
||
# 四舍五入到最近的50μs
|
||
rounded = round(pulse / 50) * 50
|
||
pulse_counts[rounded] = pulse_counts.get(rounded, 0) + 1
|
||
|
||
print(f"\n脉冲时长分布（出现次数 > 5，< 10000μs）:")
|
||
for duration in sorted(pulse_counts.keys()):
|
||
count = pulse_counts[duration]
|
||
if count > 5:
|
||
print(f" {duration}μs: {count}次")
|
||
|
||
# 找出最常见的脉冲时长
|
||
common_pulses = sorted(pulse_counts.items(), key=lambda x: x[1], reverse=True)[:10]
|
||
print(f"\n最常见的10个脉冲时长:")
|
||
for duration, count in common_pulses:
|
||
print(f" {duration}μs: {count}次")
|
||
|
||
# 分析引导码（前2个正常脉冲）
|
||
normal_start = [p for p in pulses[:10] if p < 10000]
|
||
if len(normal_start) >= 2:
|
||
print(f"\n引导码分析:")
|
||
print(f" 引导码高: {normal_start[0]}μs")
|
||
print(f" 引导码低: {normal_start[1]}μs")
|
||
|
||
# 分析数据码（跳过引导码，只看正常脉冲）
|
||
data_pulses = [p for p in pulses[2:] if p < 10000]
|
||
|
||
if len(data_pulses) > 0:
|
||
# 找出高电平脉冲（偶数位置）
|
||
high_pulses = [data_pulses[i] for i in range(0, len(data_pulses), 2)]
|
||
# 找出低电平脉冲（奇数位置）
|
||
low_pulses = [data_pulses[i] for i in range(1, len(data_pulses), 2)]
|
||
|
||
print(f"\n数据码分析:")
|
||
if high_pulses:
|
||
print(f" 数据高电平: 平均{int(statistics.mean(high_pulses))}μs, "
|
||
f"中位数{int(statistics.median(high_pulses))}μs, "
|
||
f"最小{min(high_pulses)}μs, 最大{max(high_pulses)}μs")
|
||
|
||
if low_pulses:
|
||
# 低电平有两种：短（0）和长（1）
|
||
# 使用聚类方法分离
|
||
low_pulses_sorted = sorted(low_pulses)
|
||
|
||
# 找到分界点（最大的跳跃）
|
||
max_gap = 0
|
||
split_idx = len(low_pulses_sorted) // 2
|
||
for i in range(1, len(low_pulses_sorted)):
|
||
gap = low_pulses_sorted[i] - low_pulses_sorted[i-1]
|
||
if gap > max_gap:
|
||
max_gap = gap
|
||
split_idx = i
|
||
|
||
short_lows = low_pulses_sorted[:split_idx]
|
||
long_lows = low_pulses_sorted[split_idx:]
|
||
|
||
if short_lows:
|
||
print(f" 数据0低电平: 平均{int(statistics.mean(short_lows))}μs, "
|
||
f"中位数{int(statistics.median(short_lows))}μs, "
|
||
f"范围{min(short_lows)}-{max(short_lows)}μs")
|
||
if long_lows:
|
||
print(f" 数据1低电平: 平均{int(statistics.mean(long_lows))}μs, "
|
||
f"中位数{int(statistics.median(long_lows))}μs, "
|
||
f"范围{min(long_lows)}-{max(long_lows)}μs")
|
||
|
||
def main():
|
||
files = [
|
||
'正确的csv文档/官方app制冷.csv',
|
||
'正确的csv文档/官方app制热.csv',
|
||
'正确的csv文档/官方app送风模式.csv',
|
||
'正确的csv文档/官方app抽湿.csv',
|
||
]
|
||
|
||
for filename in files:
|
||
print(f"\n{'='*60}")
|
||
print(f"分析文件: {filename}")
|
||
print(f"{'='*60}")
|
||
|
||
try:
|
||
times, states = parse_csv(filename)
|
||
pulses = calculate_pulses(times, states)
|
||
analyze_pulses(pulses)
|
||
except Exception as e:
|
||
print(f"错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == '__main__':
|
||
main()
|