448 lines
17 KiB
Python
448 lines
17 KiB
Python
import ast
|
||
from datetime import datetime, timedelta, timezone
|
||
import pandas as pd # 确保导入pandas用于类型检查
|
||
|
||
def process_incident_data(test_val):
|
||
"""
|
||
处理事故数据，去重、排序并提取指定字段。
|
||
|
||
Args:
|
||
test_val (dict): 包含 'status' 和 'data' 键的原始数据字典。
|
||
|
||
Returns:
|
||
dict: 格式化后的数据，包含 'events' 键。
|
||
"""
|
||
unique_data_dict = {}
|
||
for item in test_val:
|
||
name = item.get("incident_name")
|
||
if name and name not in unique_data_dict:
|
||
unique_data_dict[name] = item
|
||
|
||
# 获取去重后的数据列表
|
||
unique_data_list = list(unique_data_dict.values())
|
||
|
||
# 2. 根据 incident_time 升序排序
|
||
# incident_time 是 GMT 格式字符串，可以直接排序
|
||
sorted_data = sorted(unique_data_list, key=lambda x: x.get("incident_time", ""))
|
||
|
||
# 3. 提取指定字段并格式化时间
|
||
id = 0
|
||
processed_events_data = []
|
||
for item in sorted_data:
|
||
incident_time_str = item.get("incident_time")
|
||
if not incident_time_str:
|
||
# 如果 incident_time 为空或不存在，可以选择跳过或使用默认值
|
||
# 这里我们跳过
|
||
print(
|
||
f"Warning: Missing 'incident_time' for item: {item.get('incident_name')}, skipping."
|
||
)
|
||
continue
|
||
|
||
try:
|
||
# 🔥 仅修改这里：解析 GMT 格式时间 "Wed, 10 Apr 2024 16:00:00 GMT"
|
||
if isinstance(incident_time_str, datetime):
|
||
incident_datetime = incident_time_str
|
||
else:
|
||
# 解析 GMT 格式字符串 "Wed, 10 Apr 2024 16:00:00 GMT"
|
||
incident_datetime = datetime.strptime(incident_time_str, "%a, %d %b %Y %H:%M:%S %Z")
|
||
except ValueError as e:
|
||
# 如果时间格式不正确，可以选择跳过或使用默认值
|
||
# 这里我们跳过
|
||
print(
|
||
f"Warning: Invalid 'incident_time' format for item: {item.get('incident_name')}, value: {incident_time_str}, error: {e}. Skipping."
|
||
)
|
||
continue
|
||
|
||
# 格式化日期和时间（逻辑不变）
|
||
date_str = incident_datetime.strftime("%Y-%m-%d")
|
||
time_str = incident_datetime.strftime("%H:%M:%S")
|
||
|
||
# 创建新的字典，只包含需要的字段
|
||
id += 1
|
||
processed_item = {
|
||
"id": id,
|
||
"oneid": item.get("oneid", ""),
|
||
"date": date_str,
|
||
"time": time_str,
|
||
"incident_name": item.get("incident_name", ""),
|
||
"incident_description": item.get(
|
||
"incident_description", ""
|
||
),
|
||
}
|
||
processed_events_data.append(processed_item)
|
||
|
||
# 4. 构建最终返回的字典
|
||
total_count = len(processed_events_data)
|
||
result = {"data": processed_events_data, "total": total_count}
|
||
|
||
return result
|
||
|
||
|
||
def process_incident_data1(test_val):
|
||
"""
|
||
处理事故数据，按照incident_time和incident_name去重，然后按incident_time排序。
|
||
|
||
Args:
|
||
test_val (list): 原始数据列表
|
||
|
||
Returns:
|
||
dict: 格式化后的数据
|
||
"""
|
||
from datetime import datetime
|
||
|
||
# 1. 按照 incident_name 和 incident_time 组合去重
|
||
unique_data_dict = {}
|
||
for item in test_val:
|
||
name = item.get("incident_name")
|
||
time = item.get("incident_time")
|
||
if name and time:
|
||
# 使用组合键去重
|
||
key = f"{name}_{time}"
|
||
if key not in unique_data_dict:
|
||
unique_data_dict[key] = item
|
||
|
||
# 获取去重后的数据列表
|
||
unique_data_list = list(unique_data_dict.values())
|
||
|
||
# 2. 根据 incident_time 升序排序
|
||
def parse_time(time_str):
|
||
try:
|
||
# 处理 "Sat, 08 Mar 2025 09:00:00 GMT" 格式
|
||
return datetime.strptime(time_str, "%a, %d %b %Y %H:%M:%S %Z")
|
||
except:
|
||
# 处理 ISO 8601 格式作为备选
|
||
try:
|
||
return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
|
||
except:
|
||
return datetime.min
|
||
|
||
sorted_data = sorted(unique_data_list, key=lambda x: parse_time(x.get("incident_time", "")))
|
||
|
||
# 3. 提取指定字段
|
||
id = 0
|
||
processed_events_data = []
|
||
for item in sorted_data:
|
||
incident_time_str = item.get("incident_time")
|
||
if not incident_time_str:
|
||
continue
|
||
|
||
try:
|
||
incident_datetime = parse_time(incident_time_str)
|
||
except:
|
||
continue
|
||
|
||
# 格式化日期和时间
|
||
date_str = incident_datetime.strftime("%Y-%m-%d")
|
||
time_str = incident_datetime.strftime("%H:%M:%S")
|
||
|
||
# 创建新的字典
|
||
id += 1
|
||
processed_item = {
|
||
"id": id,
|
||
"oneid": item.get("oneid", ""),
|
||
"date": date_str,
|
||
"time": time_str,
|
||
"incident_name": item.get("incident_name", ""),
|
||
"incident_description": item.get("incident_description", ""),
|
||
}
|
||
processed_events_data.append(processed_item)
|
||
|
||
# 4. 构建最终返回的字典
|
||
total_count = len(processed_events_data)
|
||
result = {"data": processed_events_data, "total": total_count}
|
||
|
||
return result
|
||
|
||
def aggregate_and_sort_incidents(raw_data):
|
||
"""
|
||
按(idc_tickcount_ms + Incident__Name)聚合，整合同一组内的信号数据，消除视觉冗余
|
||
1. 同一组只输出1条记录，包含公共信息+所有信号数据
|
||
2. 按idc_tickcount_ms从小到大排序
|
||
3. 保留所有关键数据（含mux_data解析）
|
||
"""
|
||
aggregate_dict = {}
|
||
for item in raw_data:
|
||
# 1. 处理聚合key：(idc_tickcount_ms整数, Incident__Name)
|
||
try:
|
||
tickcount_int = int(item.get("idc_tickcount_ms", "0"))
|
||
except (ValueError, TypeError):
|
||
tickcount_int = 0
|
||
incident_name = item.get("incident_name", "unknown_incident")
|
||
aggregate_key = (tickcount_int, incident_name)
|
||
|
||
# 2. 提取当前记录的关键数据（用于后续整合）
|
||
# 处理数值：优先取int_value，无则取float_value，都无则为None
|
||
# value = item.get("int_value") or item.get("float_value")
|
||
int_value=item.get("int_value")
|
||
float_value=item.get('float_value')
|
||
if int_value is not None:
|
||
value = int_value
|
||
elif float_value is not None:
|
||
value = float_value
|
||
else:
|
||
value = None
|
||
|
||
# 处理mux_data：若为JSON字符串，解析为字典（方便后续使用）
|
||
mux_data = item.get("mux_data")
|
||
if mux_data and mux_data != "null":
|
||
try:
|
||
mux_data = ast.literal_eval(mux_data)
|
||
except (ValueError, SyntaxError):
|
||
pass # 解析失败则保留原始字符串
|
||
|
||
# 3. 整合到聚合字典
|
||
if aggregate_key not in aggregate_dict:
|
||
# 首次遇到该key：初始化聚合结构（提取公共基础信息）
|
||
aggregate_dict[aggregate_key] = {
|
||
"base_info": { # 同一组的公共信息（只取第一条记录的）
|
||
"oneid": item.get("oneid"),
|
||
"ihd_version": item.get("ihd_version"),
|
||
"incident_time": item.get("incident_time"),
|
||
"speed": item.get("speed"),
|
||
"gps_heading": item.get("gps_heading"),
|
||
"incident_name": incident_name,
|
||
"incident_description": item.get("incident_description"),
|
||
"Trigger__condition": item.get("Trigger__condition"),
|
||
"idc_tickcount_ms": item.get("idc_tickcount_ms"),
|
||
"idc_tickcount_ms_int": tickcount_int,
|
||
"gps_dr_position": item.get("gps_dr_position"),
|
||
"incident_id": item.get("incident_id"),
|
||
"longitude": item.get("longitude"),
|
||
"latitude": item.get("latitude")
|
||
},
|
||
"signal_data": {} # 整合同一组的所有信号（key: 信号名，value: 信号数据）
|
||
}
|
||
|
||
# print('item.get("signal_name")',item.get('signal_name'))
|
||
# 4. 将当前记录的信号数据添加到signal_data中
|
||
# signal_key = item.get("key", "unknown_key")
|
||
signal_key=item.get('signal_name')
|
||
aggregate_dict[aggregate_key]["signal_data"][signal_key] = {
|
||
"value": value,
|
||
"mux_data": mux_data,
|
||
"signal_ihd_id": item.get("signal_ihd_id"),
|
||
"Decoding__value": item.get("Decoding__value"),
|
||
"Meaning": item.get("meaning")
|
||
}
|
||
|
||
sorted_data=sort_incidents_by_heading(list(aggregate_dict.values())) # 跨周期的
|
||
|
||
sorted_result = sorted(
|
||
sorted_data,
|
||
key=lambda x: x.get("idc_tickcount_ms_tmp", x["base_info"]["idc_tickcount_ms_int"])
|
||
)
|
||
|
||
return sorted_result
|
||
|
||
|
||
def parse_coordinate(coord_str):
|
||
"""
|
||
解析坐标字符串为浮点数
|
||
"""
|
||
if coord_str is None:
|
||
return None
|
||
|
||
if isinstance(coord_str, (int, float)):
|
||
return float(coord_str)
|
||
|
||
if isinstance(coord_str, str):
|
||
try:
|
||
return float(coord_str)
|
||
except ValueError:
|
||
# 如果是 "[longitude, latitude]" 格式，提取第一个数字
|
||
if coord_str.startswith('[') and coord_str.endswith(']'):
|
||
try:
|
||
coord_str_clean = coord_str.strip('[]')
|
||
coords = coord_str_clean.split(',')
|
||
if len(coords) >= 1:
|
||
return float(coords[0].strip())
|
||
except (ValueError, IndexError):
|
||
pass
|
||
return None
|
||
|
||
return None
|
||
|
||
|
||
def sort_incidents_by_heading(raw_data):
|
||
"""
|
||
按__head方法的逻辑进行排序：根据GPS航向角排序并处理时间戳回绕
|
||
"""
|
||
import pandas as pd
|
||
|
||
# 将原始数据转换为DataFrame，并处理经纬度
|
||
processed_data = []
|
||
for item in raw_data:
|
||
# 获取base_info
|
||
base_info = item.get('base_info', {})
|
||
|
||
processed_item = item.copy() # 保持原始结构
|
||
|
||
# 解析经纬度和航向角，从base_info中获取
|
||
longitude = parse_coordinate(base_info.get('longitude'))
|
||
latitude = parse_coordinate(base_info.get('latitude'))
|
||
gps_heading = parse_coordinate(base_info.get('gps_heading'))
|
||
|
||
# 添加到processed_item，用于排序
|
||
processed_item['longitude'] = longitude if longitude is not None else 0.0
|
||
processed_item['latitude'] = latitude if latitude is not None else 0.0
|
||
processed_item['gps_heading'] = gps_heading if gps_heading is not None else 0.0
|
||
processed_item['incident_time'] = base_info.get('incident_time', '')
|
||
|
||
# 确保时间戳是数值类型，从base_info中获取
|
||
try:
|
||
tickcount_ms = int(base_info.get('idc_tickcount_ms', 0))
|
||
except (ValueError, TypeError):
|
||
tickcount_ms = 0
|
||
|
||
processed_item['idc_tickcount_ms'] = tickcount_ms
|
||
|
||
processed_data.append(processed_item)
|
||
|
||
df = pd.DataFrame(processed_data)
|
||
|
||
# 确保所有相关列都是数值类型
|
||
df['idc_tickcount_ms'] = pd.to_numeric(df['idc_tickcount_ms'], errors='coerce').fillna(0).astype(int)
|
||
df['longitude'] = pd.to_numeric(df['longitude'], errors='coerce').fillna(0.0)
|
||
df['latitude'] = pd.to_numeric(df['latitude'], errors='coerce').fillna(0.0)
|
||
df['gps_heading'] = pd.to_numeric(df['gps_heading'], errors='coerce').fillna(0.0)
|
||
|
||
# 计算平均GPS航向角
|
||
gps_heading = df["gps_heading"].mean()
|
||
# print('gps_heading',gps_heading)
|
||
|
||
# 根据GPS航向角进行排序
|
||
if (gps_heading >= 315):
|
||
df.sort_values(by=['incident_time','latitude', 'longitude','idc_tickcount_ms'],
|
||
ascending=[True, True, False, True], inplace=True)
|
||
elif (gps_heading < 45):
|
||
df.sort_values(by=['incident_time','latitude', 'longitude','idc_tickcount_ms'],
|
||
ascending=[True, True, True, True], inplace=True)
|
||
elif (gps_heading >= 45) and (gps_heading < 90):
|
||
df.sort_values(by=['incident_time','longitude', 'latitude','idc_tickcount_ms'],
|
||
ascending=[True, True, True, True], inplace=True)
|
||
elif (gps_heading >= 90) and (gps_heading < 135):
|
||
df.sort_values(by=['incident_time','longitude', 'latitude','idc_tickcount_ms'],
|
||
ascending=[True, True, False, True], inplace=True)
|
||
elif (gps_heading >= 135) and (gps_heading < 180):
|
||
df.sort_values(by=['incident_time','latitude', 'longitude','idc_tickcount_ms'],
|
||
ascending=[True, False, True, True], inplace=True)
|
||
elif (gps_heading >= 180) and (gps_heading < 225):
|
||
df.sort_values(by=['incident_time','latitude', 'longitude','idc_tickcount_ms'],
|
||
ascending=[True, False, False, True], inplace=True)
|
||
elif (gps_heading >= 225) and (gps_heading < 270):
|
||
df.sort_values(by=['incident_time','longitude', 'latitude','idc_tickcount_ms'],
|
||
ascending=[True, False, False, True], inplace=True)
|
||
elif (gps_heading >= 270) and (gps_heading < 315):
|
||
df.sort_values(by=['incident_time','longitude', 'latitude','idc_tickcount_ms'],
|
||
ascending=[True, False, True, True], inplace=True)
|
||
|
||
# 处理时间戳回绕
|
||
df['idc_tickcount_ms_tmp'] = df['idc_tickcount_ms'].copy()
|
||
df.reset_index(inplace=True, drop=True)
|
||
|
||
# 确保数据类型一致
|
||
df['idc_tickcount_ms'] = df['idc_tickcount_ms'].astype(int)
|
||
|
||
max_tickcount = df['idc_tickcount_ms'].max()
|
||
loopcount = 65535 if max_tickcount < 65536 else 2097151
|
||
period = 0
|
||
|
||
# 确保所有值都是整数
|
||
df['idc_tickcount_ms'] = df['idc_tickcount_ms'].astype(int)
|
||
|
||
for i in range(1, len(df)):
|
||
current_tick = int(df.loc[i, 'idc_tickcount_ms'])
|
||
prev_tick = int(df.loc[i-1, 'idc_tickcount_ms'])
|
||
if current_tick < prev_tick:
|
||
period += 1
|
||
df.loc[i, 'idc_tickcount_ms_tmp'] = current_tick + period * loopcount
|
||
|
||
# 按调整后的时间戳排序
|
||
df.sort_values(by=["idc_tickcount_ms_tmp"], inplace=True)
|
||
|
||
# 转换回列表格式，移除临时添加的排序字段
|
||
result = []
|
||
for record in df.to_dict('records'):
|
||
# 移除临时添加的排序字段
|
||
cleaned_record = {k: v for k, v in record.items() if k not in ['longitude', 'latitude', 'gps_heading', 'incident_time']}
|
||
result.append(cleaned_record)
|
||
|
||
# print(result)
|
||
return result
|
||
|
||
|
||
def aggregate_line_list(map_result):
|
||
# 初始化聚合字典，key为idc_tickcount_ms_int，value为聚合后的数据
|
||
aggregated_data = {}
|
||
|
||
# 遍历所有事件
|
||
for item in map_result:
|
||
base_info = item['base_info']
|
||
# 提取需要的字段
|
||
tickcount = item['idc_tickcount_ms_tmp']
|
||
speed = base_info['speed']
|
||
oneid = base_info['oneid']
|
||
latitude = base_info['latitude']
|
||
longitude = base_info['longitude']
|
||
incident_name = base_info['incident_name']
|
||
|
||
# 聚合处理
|
||
if tickcount not in aggregated_data:
|
||
# 新的tickcount，初始化记录
|
||
aggregated_data[tickcount] = {
|
||
'idc_tickcount_ms_int': tickcount,
|
||
'speed': speed,
|
||
'oneid': oneid,
|
||
'latitude': latitude,
|
||
'longitude': longitude,
|
||
'incident_names': [incident_name] # 用列表收集多个事件名称
|
||
}
|
||
else:
|
||
# 已存在的tickcount，追加事件名称
|
||
aggregated_data[tickcount]['incident_names'].append(incident_name)
|
||
|
||
# 转换为列表形式（按tickcount排序）
|
||
result = sorted(aggregated_data.values(), key=lambda x: x['idc_tickcount_ms_int'])
|
||
|
||
return result
|
||
|
||
|
||
def get_logging_list(map_list):
|
||
result=[]
|
||
id=0
|
||
for item in map_list:
|
||
# print(item)
|
||
id+=1
|
||
row_data={}
|
||
row_data['id']=id
|
||
base_info = item['base_info']
|
||
# print("base_info['incident_time']",base_info['incident_time'])
|
||
# dt = datetime.strptime(base_info['incident_time'], "%Y-%m-%d %H:%M:%S%z")
|
||
dt = base_info['incident_time']
|
||
if not isinstance(dt, datetime):
|
||
# 只有是字符串时，才解析
|
||
dt = datetime.strptime(dt, "%Y-%m-%d %H:%M:%S%z")
|
||
|
||
row_data['time']=dt.strftime("%H:%M:%S")
|
||
|
||
# row_data['coordinates']=base_info['longitude'][:12]+'.'+'\n'+base_info['latitude'][:12]
|
||
row_data['coordinates'] = str(base_info['longitude'])[:12] + '\n' + str(base_info['latitude'])[:12]
|
||
row_data['idc_tickcount_ms']=base_info['idc_tickcount_ms_int']
|
||
row_data['speed']=base_info['speed']
|
||
row_data['incident']={
|
||
"name":base_info['incident_name'],
|
||
"description":base_info['incident_description'],
|
||
}
|
||
|
||
row_data['signals']=[]
|
||
for key,value in item["signal_data"].items():
|
||
signals_name_value={}
|
||
signals_name_value["name"]=key
|
||
signals_name_value["value"]=value['value']
|
||
signals_name_value["meaning"]=value['Meaning']
|
||
row_data['signals'].append(signals_name_value)
|
||
|
||
result.append(row_data)
|
||
|
||
return result
|