如何用Python及Pandas将Excel中非结构化传感器数据转为结构化数据

核心思路 #

针对这种近似JSON但使用括号分隔的非标准格式，我们通过以下步骤处理：

1. 读取Excel文件中的原始数据列
2. 拆分原始字符串为多个逻辑区块（设备基础信息、扩展属性、时间信息）
3. 编写栈辅助的解析函数处理嵌套括号结构
4. 将解析后的结构化数据与原Excel前12列合并，输出到新Excel文件

完整代码实现 #

import pandas as pd

def parse_bracket_block(s):
    # 移除外层括号（如果存在）
    if s.startswith('[') and s.endswith(']'):
        s = s[1:-1].strip()
    
    result = {}
    current_key = None
    current_value = []
    stack = []
    
    for char in s:
        if char == '[':
            stack.append(char)
            current_value.append(char)
        elif char == ']':
            stack.pop()
            current_value.append(char)
            if not stack:
                # 嵌套区块结束，递归解析
                nested_val = parse_bracket_block(''.join(current_value))
                if current_key is not None:
                    result[current_key.strip()] = nested_val
                    current_key = None
                    current_value = []
        elif char == '=' and not stack:
            # 非嵌套状态下识别键值分隔符
            current_key = ''.join(current_value).strip()
            current_value = []
        elif char == ' ' and not stack and current_key is not None:
            # 非嵌套状态下识别值结束
            val_str = ''.join(current_value).strip()
            # 自动转换数值类型
            try:
                val = int(val_str)
            except ValueError:
                try:
                    val = float(val_str)
                except ValueError:
                    val = val_str
            result[current_key.strip()] = val
            current_key = None
            current_value = []
        else:
            current_value.append(char)
    
    # 处理剩余的键值对
    if current_key is not None and current_value:
        val_str = ''.join(current_value).strip()
        try:
            val = int(val_str)
        except ValueError:
            try:
                val = float(val_str)
            except ValueError:
                val = val_str
        result[current_key.strip()] = val
    
    return result

def parse_list_block(s):
    # 移除外层括号
    if s.startswith('[') and s.endswith(']'):
        s = s[1:-1].strip()
    
    blocks = []
    current_block = []
    stack = []
    
    for char in s:
        if char == '[':
            stack.append(char)
            current_block.append(char)
        elif char == ']':
            stack.pop()
            current_block.append(char)
            if not stack:
                blocks.append(''.join(current_block))
                current_block = []
        elif char == ',' and not stack:
            continue  # 跳过区块间的逗号分隔符
        else:
            current_block.append(char)
    
    result = {}
    for block in blocks:
        parsed = parse_bracket_block(block)
        result.update(parsed)
    
    return result

def parse_device_time(s):
    # 移除前后的#号
    s = s.lstrip('#').rstrip('#').strip()
    parts = [p.strip() for p in s.split(',')]
    result = {}
    for part in parts:
        key_val = part.split(':', 1)
        if len(key_val) == 2:
            key = key_val[0].strip()
            val = key_val[1].strip()
            result[key] = val
    return result

def parse_raw_data(raw_str):
    try:
        # 拆分主区块
        parts = raw_str.split(':::')
        if len(parts) < 2:
            return None
        
        first_section = parts[0]
        rest = parts[1]
        
        # 解析第一区块（设备类型+括号内容）
        if '[' not in first_section:
            parsed_first = {'DeviceType': first_section.strip()}
        else:
            device_type_part = first_section.split('[', 1)
            device_type = device_type_part[0].strip()
            bracket_part = '[' + device_type_part[1]
            parsed_first = parse_bracket_block(bracket_part)
            parsed_first['DeviceType'] = device_type
        
        # 拆分剩余部分为第二区块和时间区块
        rest_parts = rest.split('###')
        if len(rest_parts) <3:
            return parsed_first
        
        second_section = rest_parts[1]
        device_time_section = rest_parts[2]
        
        # 解析第二区块
        parsed_second = parse_list_block(second_section)
        
        # 解析时间区块
        parsed_time = parse_device_time(device_time_section)
        
        # 合并所有结果（后续区块覆盖重复键）
        combined = {**parsed_first, **parsed_second, **parsed_time}
        
        return combined
    except Exception as e:
        print(f"解析错误: {e}")
        return None

# 主执行逻辑
if __name__ == "__main__":
    input_file = 'sensor_data.xlsx'
    output_file = 'parsed_sensor_data.xlsx'
    
    # 读取Excel文件
    df = pd.read_excel(input_file)
    # 读取第13列（0索引为12，若为1索引则改为11）
    raw_col = df.iloc[:, 12]
    
    # 逐行解析数据
    parsed_data = []
    for raw_str in raw_col:
        if pd.isna(raw_str):
            parsed_data.append({})
            continue
        parsed = parse_raw_data(str(raw_str))
        parsed_data.append(parsed if parsed else {})
    
    # 转换为DataFrame并与原前12列合并
    parsed_df = pd.DataFrame(parsed_data)
    final_df = pd.concat([df.iloc[:, :12], parsed_df], axis=1)
    
    # 保存结果到新Excel
    final_df.to_excel(output_file, index=False)
    print(f"解析完成，结果已保存到{output_file}")

关键函数说明 #

parse_bracket_block #

处理单个括号包裹的区块（如[GPS element=[X=776517049]]），通过栈跟踪嵌套层级，自动识别嵌套结构并递归解析，同时将值转换为合适的数值类型（整数/浮点数），非数值则保留字符串。

parse_list_block #

处理逗号分隔的括号列表（如[[IMEI=357544375160179],[Latitude=128887449]]），拆分出每个独立括号区块后调用parse_bracket_block解析，最终合并为统一字典。

parse_device_time #

提取设备时间和时区信息，按逗号拆分后解析键值对。

使用步骤 #

1. 将代码保存为parse_sensor_data.py
2. 将待处理的Excel文件命名为sensor_data.xlsx放在同一目录
3. 运行脚本，解析结果会保存为parsed_sensor_data.xlsx
4. 若原始数据第13列的索引不是12（0-based），修改代码中df.iloc[:,12]的索引值

注意事项 #

• 若存在格式异常的行，脚本会打印错误信息并跳过该行解析（保留空字典）
• 重复键（如IMEI）会以后续区块的内容覆盖前面的，可根据需求调整合并逻辑
• 如需处理特殊格式数值（如十六进制），可在parse_bracket_block中添加对应类型转换逻辑

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