技能名称: json-parser

详细描述:

建筑数据JSON解析器

概述

建筑系统越来越多地使用JSON进行数据交换——从物联网传感器到BIM元数据导出。该技能处理JSON结构的解析、验证和扁平化。

Python实现

python
import json
import pandas as pd
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass
from pathlib import Path

@dataclass
class JSONParseResult:
JSON解析操作的结果。
success: bool
data: Any
errors: List[str]
record_count: int

class ConstructionJSONParser:
解析来自建筑数据源的JSON数据。

def init(self):
self.errors: List[str] = []

def parsefile(self, filepath: str) -> JSONParseResult:
从文件解析JSON。
try:
with open(file_path, r, encoding=utf-8) as f:
data = json.load(f)
return JSONParseResult(True, data, [], self.countrecords(data))
except json.JSONDecodeError as e:
return JSONParseResult(False, None, [fJSON错误: {e}], 0)
except Exception as e:
return JSONParseResult(False, None, [str(e)], 0)

def parsestring(self, jsonstring: str) -> JSONParseResult:
从字符串解析JSON。
try:
data = json.loads(json_string)
return JSONParseResult(True, data, [], self.countrecords(data))
except json.JSONDecodeError as e:
return JSONParseResult(False, None, [fJSON错误: {e}], 0)

def countrecords(self, data: Any) -> int:
统计数据中的记录数。
if isinstance(data, list):
return len(data)
elif isinstance(data, dict):
return 1
return 0

def flatten_json(self, data: Dict, prefix: str = ) -> Dict[str, Any]:
将嵌套JSON扁平化为单层字典。
flat = {}
for key, value in data.items():
newkey = f{prefix}{key} if prefix else key

if isinstance(value, dict):
flat.update(self.flattenjson(value, newkey))
elif isinstance(value, list):
if all(isinstance(i, (str, int, float, bool, type(None))) for i in value):
flat[new_key] = value
else:
for i, item in enumerate(value):
if isinstance(item, dict):
flat.update(self.flattenjson(item, f{newkey}_{i}))
else:
flat[f{newkey}{i}] = item
else:
flat[new_key] = value
return flat

def to_dataframe(self, data: Union[List[Dict], Dict]) -> pd.DataFrame:
将JSON数据转换为DataFrame。
if isinstance(data, list):
flatrecords = [self.flattenjson(r) if isinstance(r, dict) else {value: r} for r in data]
return pd.DataFrame(flat_records)
elif isinstance(data, dict):
if all(isinstance(v, list) for v in data.values()):
# 字典的列表 - 列格式
return pd.DataFrame(data)
else:
flat = self.flatten_json(data)
return pd.DataFrame([flat])
return pd.DataFrame()

def extract_elements(self, data: Dict, path: str) -> List[Any]:
使用点号表示法路径提取元素。
parts = path.split(.)
current = data

for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
elif isinstance(current, list) and part.isdigit():
current = current[int(part)]
else:
return []

return current if isinstance(current, list) else [current]

def validate_schema(self, data: Dict,
required_fields: List[str]) -> Dict[str, Any]:
根据必填字段验证JSON。
flat = self.flatten_json(data)
missing = [f for f in required_fields if f not in flat]
present = [f for f in required_fields if f in flat]

return {
valid: len(missing) == 0,
missing_fields: missing,
present_fields: present,
completeness: len(present) / len(required_fields) * 100
}

BIM JSON解析器

class BIMJSONParser(ConstructionJSONParser): 专用于BIM JSON导出的解析器。

def parsebimelements(self, data: Dict) -> pd.DataFrame:
从JSON导出解析BIM元素。
elements = []

# 常见的BIM JSON结构
if elements in data:
elements = data[elements]
elif objects in data:
elements = data[objects]
elif entities in data:
elements = data[entities]
elif isinstance(data, list):
elements = data

if not elements:
return pd.DataFrame()

# 扁平化每个元素
flat_elements = []
for elem in elements:
if isinstance(elem, dict):
flat = self.flatten_json(elem)
flat_elements.append(flat)

return pd.DataFrame(flat_elements)

def extract_properties(self, element: Dict) -> Dict[str, Any]:
从BIM元素提取属性。
props = {}

# BIM JSON中常见的属性位置
for key in [properties, params, parameters, attributes]:
if key in element and isinstance(element[key], dict):
props.update(element[key])

return props

IoT JSON解析器

class IoTJSONParser(ConstructionJSONParser): 物联网传感器数据解析器。

def parsesensorreading(self, data: Dict) -> Dict[str, Any]:
解析单个传感器读数。
return {
sensorid: data.get(sensorid) or data.get(id),
timestamp: data.get(timestamp) or data.get(time),
value: data.get(value) or data.get(reading),
unit: data.get(unit, ),
location: data.get(location, )
}

def parsesensorbatch(self, data: List[Dict]) -> pd.DataFrame:
解析批量传感器读数。
readings = [self.parsesensorreading(r) for r in data]
return pd.DataFrame(readings)

快速开始

python
parser = ConstructionJSONParser()

从文件解析

result = parser.parsefile(bimexport.json) if result.success: df = parser.to_dataframe(result.data) print(f已加载 {len(df)} 条记录)

扁平化嵌套JSON

flat = parser.flatten_json(result.data)

提取特定路径

elements = parser.extract_elements(result.data, project.building.floors)

常见用例

1. BIM元数据

python bim_parser = BIMJSONParser() result = bimparser.parsefile(revit_export.json) elements = bimparser.parsebim_elements(result.data)

2. 物联网传感器

python iot_parser = IoTJSONParser() readings = iotparser.parsesensorbatch(sensordata)

3. API响应

python parser = ConstructionJSONParser() result = parser.parsestring(apiresponse) df = parser.to_dataframe(result.data)

资源

• - DDC手册: 第2.1章 - 半结构化数据