脚本如何批量转换遥测数据格式

本文目录导读：

1. 方案一：使用 Python（最通用、推荐）
2. 方案二：使用 Bash Shell（Linux/Mac + 简单文本格式）
3. 方案三：使用 PowerShell（Windows 环境，适合 JSON/CSV 互转）
4. 方案四：处理二进制/自定义协议（FFmpeg、Protobuf 等）
5. 通用脚本模板与关键点
6. 选择建议

针对“批量转换遥测数据格式”的需求，脚本的选择取决于你的数据源格式（CSV、JSON、BIN、Protobuf等）、目标格式以及运行环境（Windows/Linux）。

以下是几种常见的批量转换方案及对应的脚本示例,你可以根据具体场景选择：

使用 Python（最通用、推荐）

Python 拥有强大的数据处理库，适合处理 CSV、JSON、XML、Parquet 等常见格式。

示例需求： 将文件夹下所有 *.csv 遥测文件转换为 *.json 格式。

脚本 convert_tlm.py：

import os
import csv
import json
import glob
def convert_csv_to_json(csv_file_path, json_file_path):
    """将单个 CSV 文件转换为 JSON 文件"""
    data = []
    try:
        with open(csv_file_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                # 可选：将字符串数字转为浮点数
                clean_row = {}
                for key, value in row.items():
                    try:
                        clean_row[key] = float(value)
                    except ValueError:
                        clean_row[key] = value
                data.append(clean_row)
        with open(json_file_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"成功转换: {csv_file_path} -> {json_file_path}")
        return True
    except Exception as e:
        print(f"转换失败 {csv_file_path}: {e}")
        return False
def batch_convert(input_dir, output_dir, input_ext='.csv', output_ext='.json'):
    """批量转换目录下所有匹配文件"""
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    # 查找所有输入文件
    pattern = os.path.join(input_dir, f'*{input_ext}')
    file_list = glob.glob(pattern)
    if not file_list:
        print(f"在 {input_dir} 下未找到 {input_ext} 文件")
        return
    success_count = 0
    for file_path in file_list:
        base_name = os.path.basename(file_path)
        name_without_ext = os.path.splitext(base_name)[0]
        output_path = os.path.join(output_dir, f'{name_without_ext}{output_ext}')
        if convert_csv_to_json(file_path, output_path):
            success_count += 1
    print(f"批量转换完成: 成功 {success_count}/{len(file_list)}")
# 使用示例
if __name__ == "__main__":
    # 配置输入输出目录
    input_directory = "./raw_telemetry/"
    output_directory = "./converted_telemetry_json/"
    batch_convert(input_directory, output_directory, input_ext='.csv', output_ext='.json')
    # 如果你想处理二进制格式，可以修改 convert 函数
    # 例如解析 .bin 文件中的自定义遥测格式

使用 Bash Shell（Linux/Mac + 简单文本格式）

如果你只需要处理简单的结构（如固定分隔符文本），且目标格式也是文本，Shell 脚本很快。

示例需求： 将 data.log 中的遥测行（空格分隔）转换为 JSON 格式。

脚本 convert_tlm.sh：

#!/bin/bash
# 配置：原始文件名的扩展名
INPUT_DIR="./raw_logs"
OUTPUT_DIR="./json_logs"
EXT=".log"
mkdir -p "$OUTPUT_DIR"
# 遍历所有 .log 文件
for file in "$INPUT_DIR"/*"$EXT"; do
    if [ -f "$file" ]; then
        base_name=$(basename "$file" "$EXT")
        output_file="$OUTPUT_DIR/$base_name.json"
        # 假设日志格式: timestamp sensor1 sensor2 sensor3
        # 输出 JSON 数组
        echo "[" > "$output_file"
        # 使用 awk 处理每行
        awk '{
            printf "{\"timestamp\":\"%s\",\"sensor1\":%s,\"sensor2\":%s,\"sensor3\":%s},\n", $1, $2, $3, $4
        }' "$file" >> "$output_file"
        # 删除最后一个逗号并关闭数组
        sed -i '$ s/,$//' "$output_file"
        echo "]" >> "$output_file"
        echo "转换完成: $file -> $output_file"
    fi
done
echo "批量转换结束"

使用 PowerShell（Windows 环境，适合 JSON/CSV 互转）

PowerShell 内置了强大的对象处理能力。

脚本 Convert-Telemetry.ps1：

# 批量将 CSV 遥测数据转换为 JSON
$InputDir = "C:\Data\RawTelemetry"
$OutputDir = "C:\Data\JsonTelemetry"
# 确保输出目录存在
New-Item -ItemType Directory -Force -Path $OutputDir
# 获取所有 CSV 文件
$files = Get-ChildItem -Path $InputDir -Filter "*.csv"
foreach ($file in $files) {
    $outputPath = Join-Path -Path $OutputDir -ChildPath ($file.BaseName + ".json")
    try {
        # 导入 CSV 并立即导出为 JSON
        Import-Csv -Path $file.FullName | ConvertTo-Json -Depth 5 | Out-File -FilePath $outputPath -Encoding UTF8
        Write-Host "转换成功: $($file.Name)"
    }
    catch {
        Write-Warning "转换失败: $($file.Name) - $_"
    }
}
Write-Host "批量转换完成。"

处理二进制/自定义协议（FFmpeg、Protobuf 等）

如果你的遥测数据是二进制格式（如传感器厂商自定义协议）,你需要使用专门的解析器。

示例：使用 struct 模块解析二进制遥测（Python）：

import os
import struct
import json
def convert_bin_to_json(bin_path, json_path):
    """假设每个数据包为 20 字节: uint32 timestamp, float x, float y, float z, uint16 status"""
    records = []
    with open(bin_path, 'rb') as f:
        while True:
            chunk = f.read(20)
            if len(chunk) < 20:
                break
            # 解包 (小端格式)
            timestamp, x, y, z, status = struct.unpack('<IfffH', chunk)
            records.append({
                'timestamp': timestamp,
                'accel_x': x,
                'accel_y': y,
                'accel_z': z,
                'status': status
            })
    with open(json_path, 'w') as f:
        json.dump(records, f, indent=2)
# 批量转换
for bin_file in glob.glob('./raw_bin/*.bin'):
    output_name = os.path.splitext(os.path.basename(bin_file))[0] + '.json'
    convert_bin_to_json(bin_file, f'./output_json/{output_name}')

通用脚本模板与关键点

无论使用哪种脚本,核心逻辑都是以下几步：

1. 遍历文件： 使用 glob、os.listdir 或 Shell 通配符查找所有待处理文件。
2. 打开与读取： 根据原始格式确定读取方式（csv、json.load、二进制 struct）。
3. 数据清洗与转换：
   • 处理缺失值（NaN、null）。
   • 时间戳格式统一（Unix 时间戳 ↔ 可读字符串）。
   • 单位换算（如 int32 原始值 → 物理量）。
4. 写入目标格式： 确保编码一致（推荐 UTF-8）,目标目录存在。
5. 错误处理： 使用 try-except 或 if-else 跳过单个坏文件,不影响整体流程。

选择建议

你可以根据你的具体原始格式（CSV？二进制？）和目标格式（JSON？Parquet？）修改上述脚本中的核心转换函数，如果需要更具体的帮助,请提供你的数据样例文件和目标格式要求。