# ... 保留原有导入和常量定义 ... import math import operator import copy import numpy as np import pandas as pd from typing import Dict, Any, List, Optional from modules.lib import log_manager from modules.lib.score import Score from modules.lib.log_manager import LogManager from modules.lib import data_process OVERTAKE_INFO = [ "simTime", "simFrame", "playerId", "speedX", "speedY", "posX", "posY", "posH", "lane_id", "lane_type", "road_type", "interid", "crossid", ] SLOWDOWN_INFO = [ "simTime", "simFrame", "playerId", "speedX", "speedY", "posX", "posY", "crossid", "lane_type", ] TURNAROUND_INFO = [ "simTime", "simFrame", "playerId", "speedX", "speedY", "posX", "posY", "sign_type1", "lane_type", ] TRFFICSIGN_INFO = [ "simTime", "simFrame", "playerId", "speedX", "speedY", "v", "posX", "posY", "sign_type1", "sign_ref_link", "sign_x", "sign_y", ] # 修改指标函数名称为 calculate_xxx 格式 def calculate_overtake_when_passing_car(data_processed): """计算会车时超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_when_passing_car_count = overtakingviolation.calculate_overtake_when_passing_car_count() return {"overtake_when_passing_car": overtake_when_passing_car_count} def calculate_overtake_on_right(data_processed): """计算右侧超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_on_right_count = overtakingviolation.calculate_overtake_on_right_count() return {"overtake_on_right": overtake_on_right_count} def calculate_overtake_when_turn_around(data_processed): """计算掉头时超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_when_turn_around_count = overtakingviolation.calculate_overtake_when_turn_around_count() return {"overtake_when_turn_around": overtake_when_turn_around_count} def calculate_overtake_in_forbid_lane(data_processed): """计算在禁止车道超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_in_forbid_lane_count = overtakingviolation.calculate_overtake_in_forbid_lane_count() return {"overtake_in_forbid_lane": overtake_in_forbid_lane_count} def calculate_overtake_in_ramp(data_processed): """计算在匝道超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_in_ramp_area_count = overtakingviolation.calculate_overtake_in_ramp_area_count() return {"overtake_in_ramp": overtake_in_ramp_area_count} def calculate_overtake_in_tunnel(data_processed): """计算在隧道超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_in_tunnel_area_count = overtakingviolation.calculate_overtake_in_tunnel_area_count() return {"overtake_in_tunnel": overtake_in_tunnel_area_count} def calculate_overtake_on_accelerate_lane(data_processed): """计算在加速车道超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_on_accelerate_lane_count = overtakingviolation.calculate_overtake_on_accelerate_lane_count() return {"overtake_on_accelerate_lane": overtake_on_accelerate_lane_count} def calculate_overtake_on_decelerate_lane(data_processed): """计算在减速车道超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_on_decelerate_lane_count = overtakingviolation.calculate_overtake_on_decelerate_lane_count() return {"overtake_on_decelerate_lane": overtake_on_decelerate_lane_count} def calculate_overtake_in_different_senerios(data_processed): """计算在不同场景超车指标""" overtakingviolation = OvertakingViolation(data_processed) overtake_in_different_senerios_count = overtakingviolation.calculate_overtake_in_different_senerios_count() return {"overtake_in_different_senerios": overtake_in_different_senerios_count} def calculate_slow_down_in_crosswalk(data_processed): """计算在人行横道减速指标""" slowdownviolation = SlowdownViolation(data_processed) slow_down_in_crosswalk_count = slowdownviolation.calculate_slow_down_in_crosswalk_count() return {"slowdown_down_in_crosswalk": slow_down_in_crosswalk_count} def calculate_avoid_pedestrian_in_crosswalk(data_processed): """计算在人行横道避让行人指标""" avoidpedestrianincrosswalk = SlowdownViolation(data_processed) avoid_pedestrian_in_crosswalk_count = avoidpedestrianincrosswalk.calculate_avoid_pedestrian_in_the_crosswalk_count() return {"avoid_pedestrian_in_crosswalk": avoid_pedestrian_in_crosswalk_count} def calculate_avoid_pedestrian_in_the_road(data_processed): """计算在道路上避让行人指标""" avoidpedestrianintheroad = SlowdownViolation(data_processed) avoid_pedestrian_in_the_road_count = avoidpedestrianintheroad.calculate_avoid_pedestrian_in_the_road_count() return {"avoid_pedestrian_in_the_road": avoid_pedestrian_in_the_road_count} def calculate_avoid_pedestrian_when_turning(data_processed): """计算转弯时避让行人指标""" avoidpedestrianwhenturning = SlowdownViolation(data_processed) avoid_pedestrian_when_turning_count = avoidpedestrianwhenturning.calculate_avoid_pedestrian_when_turning_count() return {"avoid_pedestrian_when_turning_count": avoid_pedestrian_when_turning_count} def calculate_turn_in_forbiden_turn_left_sign(data_processed): """计算在禁止左转标志处左转指标""" turnaroundviolation = TurnaroundViolation(data_processed) turn_in_forbiden_turn_left_sign_count = turnaroundviolation.calculate_turn_in_forbiden_turn_left_sign_count() return {"turn_in_forbiden_turn_left_sign": turn_in_forbiden_turn_left_sign_count} def calculate_turn_in_forbiden_turn_back_sign(data_processed): """计算在禁止掉头标志处掉头指标""" turnaroundviolation = TurnaroundViolation(data_processed) turn_in_forbiden_turn_back_sign_count = turnaroundviolation.calculate_turn_in_forbiden_turn_back_sign_count() return {"turn_in_forbiden_turn_back_sign": turn_in_forbiden_turn_back_sign_count} def calculate_avoid_pedestrian_when_turn_back(data_processed): """计算掉头时避让行人指标""" turnaroundviolation = TurnaroundViolation(data_processed) avoid_pedestrian_when_turn_back_count = turnaroundviolation.calaulate_avoid_pedestrian_when_turn_back_count() return {"avoid_pedestrian_when_turn_back": avoid_pedestrian_when_turn_back_count} def calculate_urbanexpresswayorhighwaydrivinglanestopped(data_processed): """计算城市快速路或高速公路行车道停车指标""" wrongwayviolation = WrongWayViolation(data_processed) urbanExpresswayOrHighwayDrivingLaneStopped_count = wrongwayviolation.calculate_urbanExpresswayOrHighwayDrivingLaneStopped_count() return {"urbanExpresswayOrHighwayDrivingLaneStopped": urbanExpresswayOrHighwayDrivingLaneStopped_count} def calculate_urbanexpresswayorhighwayemergencylanestopped(data_processed): """计算城市快速路或高速公路应急车道停车指标""" wrongwayviolation = WrongWayViolation(data_processed) urbanExpresswayOrHighwayEmergencyLaneStopped_count = wrongwayviolation.calculate_urbanExpresswayOrHighwayDrivingLaneStopped_count() return {"urbanExpresswayOrHighwayEmergencyLaneStopped": urbanExpresswayOrHighwayEmergencyLaneStopped_count} def calculate_urbanexpresswayemergencylanedriving(data_processed): """计算城市快速路应急车道行驶指标""" wrongwayviolation = WrongWayViolation(data_processed) urbanExpresswayEmergencyLaneDriving_count = wrongwayviolation.calculate_urbanExpresswayEmergencyLaneDriving() return {"urbanExpresswayEmergencyLaneDriving": urbanExpresswayEmergencyLaneDriving_count} def calculate_urbanexpresswayorhighwayspeedoverlimit50(data_processed): """计算城市快速路或高速公路超速50%以上指标""" speedingviolation = SpeedingViolation(data_processed) urbanExpresswayOrHighwaySpeedOverLimit50_count = speedingviolation.calculate_urbanExpresswayOrHighwaySpeedOverLimit50_count() return {"urbanExpresswayOrHighwaySpeedOverLimit50": urbanExpresswayOrHighwaySpeedOverLimit50_count} def calculate_urbanexpresswayorhighwayspeedoverlimit20to50(data_processed): """计算城市快速路或高速公路超速20%-50%指标""" speedingviolation = SpeedingViolation(data_processed) urbanExpresswayOrHighwaySpeedOverLimit20to50_count = speedingviolation.calculate_urbanExpresswayOrHighwaySpeedOverLimit20to50_count() return {"urbanExpresswayOrHighwaySpeedOverLimit20to50": urbanExpresswayOrHighwaySpeedOverLimit20to50_count} def calculate_urbanexpresswayorhighwayspeedoverlimit0to20(data_processed): """计算城市快速路或高速公路超速0-20%指标""" speedingviolation = SpeedingViolation(data_processed) urbanExpresswayOrHighwaySpeedOverLimit0to20_count = speedingviolation.calculate_urbanExpresswayOrHighwaySpeedOverLimit0to20_count() return {"urbanExpresswayOrHighwaySpeedOverLimit0to20": urbanExpresswayOrHighwaySpeedOverLimit0to20_count} def calculate_urbanexpresswayorhighwayspeedunderlimit(data_processed): """计算城市快速路或高速公路低于最低限速指标""" speedingviolation = SpeedingViolation(data_processed) urbanExpresswayOrHighwaySpeedUnderLimit_count = speedingviolation.calculate_urbanExpresswayOrHighwaySpeedUnderLimit_count() return {"urbanExpresswayOrHighwaySpeedUnderLimit": urbanExpresswayOrHighwaySpeedUnderLimit_count} def calculate_generalroadspeedoverlimit50(data_processed): """计算一般道路超速50%以上指标""" speedingviolation = SpeedingViolation(data_processed) generalRoadSpeedOverLimit50_count = speedingviolation.calculate_generalRoadSpeedOverLimit50() return {"generalRoadSpeedOverLimit50": generalRoadSpeedOverLimit50_count} def calculate_generalroadspeedoverlimit20to50(data_processed): """计算一般道路超速20%-50%指标""" speedingviolation = SpeedingViolation(data_processed) generalRoadSpeedOverLimit20to50_count = speedingviolation.calculate_generalRoadSpeedOverLimit20to50_count() return {"generalRoadSpeedOverLimit20to50": generalRoadSpeedOverLimit20to50_count} def calculate_trafficsignalviolation(data_processed): """计算交通信号违规指标""" trafficlightviolation = TrafficLightViolation(data_processed) trafficSignalViolation_count = trafficlightviolation.calculate_trafficSignalViolation_count() return {"trafficSignalViolation": trafficSignalViolation_count} def calculate_illegaldrivingorparkingatcrossroads(data_processed): """计算交叉路口违法行驶或停车指标""" trafficlightviolation = TrafficLightViolation(data_processed) illegalDrivingOrParkingAtCrossroads_count = trafficlightviolation.calculate_illegalDrivingOrParkingAtCrossroads() return {"illegalDrivingOrParkingAtCrossroads": illegalDrivingOrParkingAtCrossroads_count} def calculate_generalroadirregularlaneuse(data_processed): """计算一般道路不按规定车道行驶指标""" warningviolation = WarningViolation(data_processed) generalRoadIrregularLaneUse_count = warningviolation.calculate_generalRoadIrregularLaneUse_count() return {"generalRoadIrregularLaneUse": generalRoadIrregularLaneUse_count} def calculate_urbanexpresswayorhighwayridelanedivider(data_processed): """计算城市快速路或高速公路骑车道线行驶指标""" warningviolation = WarningViolation(data_processed) urbanExpresswayOrHighwayRideLaneDivider_count = warningviolation.calculate_urbanExpresswayOrHighwayRideLaneDivider_count() return {"urbanExpresswayOrHighwayRideLaneDivider": urbanExpresswayOrHighwayRideLaneDivider_count} def calculate_nostraightthrough(data_processed): """计算禁止直行标志牌处直行指标""" trafficsignviolation = TrafficSignViolation(data_processed) noStraightThrough_count = trafficsignviolation.calculate_NoStraightThrough_count() return {"NoStraightThrough": noStraightThrough_count} def calculate_speedlimitviolation(data_processed): """计算违反限速规定指标""" trafficsignviolation = TrafficSignViolation(data_processed) SpeedLimitViolation_count = trafficsignviolation.calculate_SpeedLimitViolation_count() return {"SpeedLimitViolation": SpeedLimitViolation_count} def calculate_minimumspeedlimitviolation(data_processed): """计算违反最低限速规定指标""" trafficsignviolation = TrafficSignViolation(data_processed) calculate_MinimumSpeedLimitViolation_count = trafficsignviolation.calculate_MinimumSpeedLimitViolation_count() return {"MinimumSpeedLimitViolation": calculate_MinimumSpeedLimitViolation_count} # 修改 TrafficRegistry 类的 _build_registry 方法 class TrafficRegistry: """交通违规指标注册器""" def __init__(self, data_processed): self.logger = LogManager().get_logger() self.data = data_processed self.traffic_config = data_processed.traffic_config["traffic"] self.metrics = self._extract_metrics(self.traffic_config) self._registry = self._build_registry() def _extract_metrics(self, config_node: dict) -> list: """从配置中提取指标名称""" metrics = [] def _recurse(node): if isinstance(node, dict): if 'name' in node and not any(isinstance(v, dict) for v in node.values()): metrics.append(node['name']) for v in node.values(): _recurse(v) _recurse(config_node) self.logger.info(f'评比的交通违规指标列表:{metrics}') return metrics def _build_registry(self) -> dict: """构建指标函数注册表""" registry = {} for metric_name in self.metrics: func_name = f"calculate_{metric_name.lower()}" try: registry[metric_name] = globals()[func_name] except KeyError: self.logger.error(f"未实现交通违规指标函数: {func_name}") return registry def batch_execute(self) -> dict: """批量执行指标计算""" results = {} for name, func in self._registry.items(): try: result = func(self.data) results.update(result) # 新增：将每个指标的结果写入日志 self.logger.info(f'交通违规指标[{name}]计算结果: {result}') except Exception as e: self.logger.error(f"{name} 执行失败: {str(e)}", exc_info=True) results[name] = None self.logger.info(f'交通违规指标计算结果:{results}') return results class TrafficManager: """交通违规指标管理类""" def __init__(self, data_processed): self.data = data_processed self.logger = LogManager().get_logger() self.registry = TrafficRegistry(self.data) def report_statistic(self): """计算并报告交通违规指标结果""" traffic_result = self.registry.batch_execute() return traffic_result class OvertakingViolation(object): """超车违规类""" def __init__(self, df_data): print("超车违规类初始化中...") self.traffic_violations_type = "超车违规类" # 存储原始数据引用，不进行拷贝 self._raw_data = df_data.obj_data[1] # 自车数据 # 安全获取其他车辆数据 self._data_obj = None self._other_obj_data1 = None # 检查是否存在ID为2的对象数据 if 2 in df_data.obj_id_list: self._data_obj = df_data.obj_data[2] # 检查是否存在ID为3的对象数据 if 3 in df_data.obj_id_list: self._other_obj_data1 = df_data.obj_data[3] # 初始化属性，但不立即创建数据副本 self._ego_data = None self._obj_data = None self._other_obj_data = None # 使用字典统一管理违规计数器 self.violation_counts = { "overtake_on_right": 0, "overtake_when_turn_around": 0, "overtake_when_passing_car": 0, "overtake_in_forbid_lane": 0, "overtake_in_ramp": 0, "overtake_in_tunnel": 0, "overtake_on_accelerate_lane": 0, "overtake_on_decelerate_lane": 0, "overtake_in_different_senerios": 0 } # 标记计算状态 self._calculated = { "illegal_overtake": False, "forbid_lane": False, "ramp_area": False, "tunnel_area": False, "accelerate_lane": False, "decelerate_lane": False, "different_senerios": False } @property def ego_data(self): """懒加载方式获取ego数据，只在首次访问时创建副本""" if self._ego_data is None: self._ego_data = self._raw_data[OVERTAKE_INFO].copy().reset_index(drop=True) return self._ego_data @property def obj_data(self): """懒加载方式获取obj数据""" if self._obj_data is None: if self._data_obj is not None: self._obj_data = self._data_obj[OVERTAKE_INFO].copy().reset_index(drop=True) else: # 如果没有数据，创建一个空的DataFrame，列名与ego_data相同 self._obj_data = pd.DataFrame(columns=OVERTAKE_INFO) return self._obj_data @property def other_obj_data(self): """懒加载方式获取other_obj数据""" if self._other_obj_data is None: if self._other_obj_data1 is not None: self._other_obj_data = self._other_obj_data1[OVERTAKE_INFO].copy().reset_index(drop=True) else: # 如果没有数据，创建一个空的DataFrame，列名与ego_data相同 self._other_obj_data = pd.DataFrame(columns=OVERTAKE_INFO) return self._other_obj_data def different_road_area_simtime(self, df, threshold=0.5): if not df: return [] simtime_group = [] current_simtime_group = [df[0]] for i in range(1, len(df)): if abs(df[i] - df[i - 1]) <= threshold: current_simtime_group.append(df[i]) else: simtime_group.append(current_simtime_group) current_simtime_group = [df[i]] simtime_group.append(current_simtime_group) return simtime_group def _is_overtake(self, lane_id, dx, dy, ego_speedx, ego_speedy): lane_start = lane_id[0] lane_end = lane_id[-1] start_condition = dx[0] * ego_speedx[0] + dy[0] * ego_speedy[0] >= 0 end_condition = dx[-1] * ego_speedx[-1] + dy[-1] * ego_speedy[-1] < 0 return lane_start == lane_end and start_condition and end_condition def _is_dxy_of_car(self, ego_df, obj_df): """ :param df: objstate.csv and so on :param id: playerId :param string_type: posX/Y or speedX/Y and so on :return: dataframe of dx/y and so on """ car_dx = obj_df["posX"].values - ego_df["posX"].values car_dy = obj_df["posY"].values - ego_df["posY"].values return car_dx, car_dy def illegal_overtake_with_car_detector(self, window_width=250): """检测超车违规""" # 如果已经计算过，直接返回 if self._calculated["illegal_overtake"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测超车违规，默认为0") self._calculated["illegal_overtake"] = True return # 获取csv文件中最短的帧数 frame_id_length = len(self.ego_data["simFrame"]) start_frame_id = self.ego_data["simFrame"].iloc[0] # 获取起始点的帧数 while (start_frame_id + window_width) < frame_id_length: simframe_window1 = list( np.arange(start_frame_id, start_frame_id + window_width) ) simframe_window = list(map(int, simframe_window1)) # 读取滑动窗口的dataframe数据 ego_data_frames = self.ego_data[ self.ego_data["simFrame"].isin(simframe_window) ] # 确保有足够的数据进行处理 if len(ego_data_frames) == 0: start_frame_id += 1 continue obj_data_frames = self.obj_data[ self.obj_data["simFrame"].isin(simframe_window) ] # 如果没有其他车辆数据，跳过当前窗口 if len(obj_data_frames) == 0: start_frame_id += 1 continue other_data_frames = self.other_obj_data[ self.other_obj_data["simFrame"].isin(simframe_window) ] # 读取前后的laneId lane_id = ego_data_frames["lane_id"].tolist() # 读取前后方向盘转角steeringWheel driverctrl_start_state = ego_data_frames["posH"].iloc[0] driverctrl_end_state = ego_data_frames["posH"].iloc[-1] # 读取车辆前后的位置信息 dx, dy = self._is_dxy_of_car(ego_data_frames, obj_data_frames) ego_speedx = ego_data_frames["speedX"].tolist() ego_speedy = ego_data_frames["speedY"].tolist() # 安全获取obj_speedx和obj_speedy obj_with_id_2 = obj_data_frames[obj_data_frames["playerId"] == 2] if not obj_with_id_2.empty: obj_speedx = obj_with_id_2["speedX"].tolist() obj_speedy = obj_with_id_2["speedY"].tolist() else: obj_speedx = [] obj_speedy = [] # 检查会车时超车 if len(other_data_frames) > 0: other_start_speedx = other_data_frames["speedX"].iloc[0] other_start_speedy = other_data_frames["speedY"].iloc[0] if ( ego_speedx[0] * other_start_speedx + ego_speedy[0] * other_start_speedy < 0 ): self.violation_counts["overtake_when_passing_car"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) start_frame_id += window_width continue # 检查右侧超车 if driverctrl_start_state > 0 and driverctrl_end_state < 0: self.violation_counts["overtake_on_right"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) start_frame_id += window_width continue # 检查掉头时超车 if obj_speedx and obj_speedy: # 确保列表不为空 if ego_speedx[0] * obj_speedx[0] + ego_speedy[0] * obj_speedy[0] < 0: self.violation_counts["overtake_when_turn_around"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) start_frame_id += window_width continue # 如果没有检测到任何违规，移动窗口 start_frame_id += 1 self._calculated["illegal_overtake"] = True # 借道超车场景 def overtake_in_forbid_lane_detector(self): """检测借道超车违规""" # 如果已经计算过，直接返回 if self._calculated["forbid_lane"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测借道超车违规，默认为0") self._calculated["forbid_lane"] = True return simTime = self.obj_data["simTime"].tolist() simtime_devide = self.different_road_area_simtime(simTime) for simtime in simtime_devide: lane_overtake = self.ego_data[self.ego_data["simTime"].isin(simtime)] try: lane_type = lane_overtake["lane_type"].tolist() if (50002 in lane_type and len(set(lane_type)) > 2) or ( 50002 not in lane_type and len(set(lane_type)) > 1 ): self.violation_counts["overtake_in_forbid_lane"] += 1 except Exception as e: print("数据缺少lane_type信息") self._calculated["forbid_lane"] = True # 在匝道超车 def overtake_in_ramp_area_detector(self): """检测匝道超车违规""" # 如果已经计算过，直接返回 if self._calculated["ramp_area"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测匝道超车违规，默认为0") self._calculated["ramp_area"] = True return ramp_simtime_list = self.ego_data[(self.ego_data["road_type"] == 19)][ "simTime" ].tolist() ramp_simTime_list = self.different_road_area_simtime(ramp_simtime_list) for ramp_simtime in ramp_simTime_list: lane_id = self.ego_data["lane_id"].tolist() ego_in_ramp = self.ego_data[self.ego_data["simTime"].isin(ramp_simtime)] objstate_in_ramp = self.obj_data[ self.obj_data["simTime"].isin(ramp_simtime) ] dx, dy = self._is_dxy_of_car(ego_in_ramp, objstate_in_ramp) ego_speedx = ego_in_ramp["speedX"].tolist() ego_speedy = ego_in_ramp["speedY"].tolist() if len(lane_id) > 0: self.violation_counts["overtake_in_ramp"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) else: continue self._calculated["ramp_area"] = True def overtake_in_tunnel_area_detector(self): """检测隧道超车违规""" # 如果已经计算过，直接返回 if self._calculated["tunnel_area"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测隧道超车违规，默认为0") self._calculated["tunnel_area"] = True return tunnel_simtime_list = self.ego_data[(self.ego_data["road_type"] == 15)][ "simTime" ].tolist() tunnel_simTime_list = self.different_road_area_simtime(tunnel_simtime_list) for tunnel_simtime in tunnel_simTime_list: lane_id = self.ego_data["lane_id"].tolist() ego_in_tunnel = self.ego_data[self.ego_data["simTime"].isin(tunnel_simtime)] objstate_in_tunnel = self.obj_data[ self.obj_data["simTime"].isin(tunnel_simtime) ] dx, dy = self._is_dxy_of_car(ego_in_tunnel, objstate_in_tunnel) ego_speedx = ego_in_tunnel["speedX"].tolist() ego_speedy = ego_in_tunnel["speedY"].tolist() if len(lane_id) > 0: self.violation_counts["overtake_in_tunnel"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) else: continue self._calculated["tunnel_area"] = True # 加速车道超车 def overtake_on_accelerate_lane_detector(self): """检测加速车道超车违规""" # 如果已经计算过，直接返回 if self._calculated["accelerate_lane"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测加速车道超车违规，默认为0") self._calculated["accelerate_lane"] = True return accelerate_simtime_list = self.ego_data[self.ego_data["lane_type"] == 2][ "simTime" ].tolist() accelerate_simTime_list = self.different_road_area_simtime( accelerate_simtime_list ) for accelerate_simtime in accelerate_simTime_list: lane_id = self.ego_data["lane_id"].tolist() ego_in_accelerate = self.ego_data[ self.ego_data["simTime"].isin(accelerate_simtime) ] objstate_in_accelerate = self.obj_data[ self.obj_data["simTime"].isin(accelerate_simtime) ] dx, dy = self._is_dxy_of_car(ego_in_accelerate, objstate_in_accelerate) ego_speedx = ego_in_accelerate["speedX"].tolist() ego_speedy = ego_in_accelerate["speedY"].tolist() self.violation_counts["overtake_on_accelerate_lane"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) self._calculated["accelerate_lane"] = True # 减速车道超车 def overtake_on_decelerate_lane_detector(self): """检测减速车道超车违规""" # 如果已经计算过，直接返回 if self._calculated["decelerate_lane"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测减速车道超车违规，默认为0") self._calculated["decelerate_lane"] = True return decelerate_simtime_list = self.ego_data[(self.ego_data["lane_type"] == 3)][ "simTime" ].tolist() decelerate_simTime_list = self.different_road_area_simtime( decelerate_simtime_list ) for decelerate_simtime in decelerate_simTime_list: lane_id = self.ego_data["id"].tolist() ego_in_decelerate = self.ego_data[ self.ego_data["simTime"].isin(decelerate_simtime) ] objstate_in_decelerate = self.obj_data[ self.obj_data["simTime"].isin(decelerate_simtime) ] dx, dy = self._is_dxy_of_car(ego_in_decelerate, objstate_in_decelerate) ego_speedx = ego_in_decelerate["speedX"].tolist() ego_speedy = ego_in_decelerate["speedY"].tolist() self.violation_counts["overtake_on_decelerate_lane"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) self._calculated["decelerate_lane"] = True # 在交叉路口 def overtake_in_different_senerios_detector(self): """检测不同场景超车违规""" # 如果已经计算过，直接返回 if self._calculated["different_senerios"]: return # 如果没有其他车辆数据，直接返回，保持默认值0 if self.obj_data.empty: print("没有其他车辆数据，无法检测不同场景超车违规，默认为0") self._calculated["different_senerios"] = True return crossroad_simTime = self.ego_data[self.ego_data["interid"] != 10000][ "simTime" ].tolist() # 判断是路口或者隧道区域 # 筛选在路口或者隧道区域的objectstate、driverctrl、laneinfo数据 crossroad_ego = self.ego_data[self.ego_data["simTime"].isin(crossroad_simTime)] crossroad_objstate = self.obj_data[ self.obj_data["simTime"].isin(crossroad_simTime) ] # 读取前后的laneId lane_id = crossroad_ego["lane_id"].tolist() # 读取车辆前后的位置信息 dx, dy = self._is_dxy_of_car(crossroad_ego, crossroad_objstate) ego_speedx = crossroad_ego["speedX"].tolist() ego_speedy = crossroad_ego["speedY"].tolist() """ 如果滑动窗口开始和最后的laneid一致； 自车和前车的位置发生的交换； 则认为发生超车 """ if len(lane_id) > 0: self.violation_counts["overtake_in_different_senerios"] += self._is_overtake( lane_id, dx, dy, ego_speedx, ego_speedy ) self._calculated["different_senerios"] = True def calculate_overtake_when_passing_car_count(self): """计算会车时超车违规次数""" self.illegal_overtake_with_car_detector() return self.violation_counts["overtake_when_passing_car"] def calculate_overtake_on_right_count(self): """计算右侧超车违规次数""" self.illegal_overtake_with_car_detector() return self.violation_counts["overtake_on_right"] def calculate_overtake_when_turn_around_count(self): """计算掉头时超车违规次数""" self.illegal_overtake_with_car_detector() return self.violation_counts["overtake_when_turn_around"] def calculate_overtake_in_forbid_lane_count(self): """计算借道超车违规次数""" self.overtake_in_forbid_lane_detector() return self.violation_counts["overtake_in_forbid_lane"] def calculate_overtake_in_ramp_area_count(self): """计算匝道超车违规次数""" self.overtake_in_ramp_area_detector() return self.violation_counts["overtake_in_ramp"] def calculate_overtake_in_tunnel_area_count(self): """计算隧道超车违规次数""" self.overtake_in_tunnel_area_detector() return self.violation_counts["overtake_in_tunnel"] def calculate_overtake_on_accelerate_lane_count(self): """计算加速车道超车违规次数""" self.overtake_on_accelerate_lane_detector() return self.violation_counts["overtake_on_accelerate_lane"] def calculate_overtake_on_decelerate_lane_count(self): """计算减速车道超车违规次数""" self.overtake_on_decelerate_lane_detector() return self.violation_counts["overtake_on_decelerate_lane"] def calculate_overtake_in_different_senerios_count(self): """计算不同场景超车违规次数""" self.overtake_in_different_senerios_detector() return self.violation_counts["overtake_in_different_senerios"] class SlowdownViolation(object): """减速让行违规类""" def __init__(self, df_data): print("减速让行违规类-------------------------") self.traffic_violations_type = "减速让行违规类" # 存储原始数据引用 self._raw_data = df_data.obj_data[1] self.object_items = set(df_data.object_df.type.tolist()) # 存储行人数据引用 self._pedestrian_df = None if 13 in self.object_items: # 行人的type是13 self._pedestrian_df = df_data.object_df[df_data.object_df.type == 13] # 初始化属性，但不立即创建数据副本 self._ego_data = None self._pedestrian_data = None # 初始化计数器 self.slow_down_in_crosswalk_count = 0 self.avoid_pedestrian_in_crosswalk_count = 0 self.avoid_pedestrian_in_the_road_count = 0 self.aviod_pedestrian_when_turning_count = 0 @property def ego_data(self): """懒加载方式获取ego数据""" if self._ego_data is None: self._ego_data = self._raw_data[SLOWDOWN_INFO].copy().reset_index(drop=True) return self._ego_data @property def pedestrian_data(self): """懒加载方式获取行人数据""" if self._pedestrian_data is None: if self._pedestrian_df is not None: # 使用浅拷贝代替深拷贝 self._pedestrian_data = self._pedestrian_df[SLOWDOWN_INFO].copy().reset_index(drop=True) else: self._pedestrian_data = pd.DataFrame() return self._pedestrian_data def pedestrian_in_front_of_car(self): if len(self.pedestrian_data) == 0: return [] else: self.ego_data["dx"] = self.ego_data["posX"] - self.pedestrian_data["posX"] self.ego_data["dy"] = self.ego_data["posY"] - self.pedestrian_data["posY"] self.ego_data["dist"] = np.sqrt( self.ego_data["dx"] ** 2 + self.ego_data["dy"] ** 2 ) self.ego_data["rela_pos"] = ( self.ego_data["dx"] * self.ego_data["speedX"] + self.ego_data["dy"] * self.ego_data["speedY"] ) simtime = self.ego_data[ (self.ego_data["rela_pos"] > 0) & (self.ego_data["dist"] < 50) ]["simTime"].tolist() return simtime def different_road_area_simtime(self, df, threshold=0.6): if not df: return [] simtime_group = [] current_simtime_group = [df[0]] for i in range(1, len(df)): if abs(df[i] - df[i - 1]) <= threshold: current_simtime_group.append(df[i]) else: simtime_group.append(current_simtime_group) current_simtime_group = [df[i]] simtime_group.append(current_simtime_group) return simtime_group def slow_down_in_crosswalk_detector(self): # 筛选出路口或隧道区域的时间点 crosswalk_simTime = self.ego_data[self.ego_data["crossid"] != 20000][ "simTime" ].tolist() crosswalk_simTime_divide = self.different_road_area_simtime(crosswalk_simTime) for crosswalk_simtime in crosswalk_simTime_divide: # 筛选出当前时间段内的数据 # start_time, end_time = crosswalk_simtime start_time = crosswalk_simtime[0] end_time = crosswalk_simtime[-1] print(f"当前时间段：{start_time} - {end_time}") crosswalk_objstate = self.ego_data[ (self.ego_data["simTime"] >= start_time) & (self.ego_data["simTime"] <= end_time) ] # 计算车辆速度 ego_speedx = np.array(crosswalk_objstate["speedX"].tolist()) ego_speedy = np.array(crosswalk_objstate["speedY"].tolist()) ego_speed = np.sqrt(ego_speedx ** 2 + ego_speedy ** 2) # 判断是否超速 if max(ego_speed) > 15 / 3.6: # 15 km/h 转换为 m/s self.slow_down_in_crosswalk_count += 1 # 输出总次数 print(f"在人行横道超车总次数：{self.slow_down_in_crosswalk_count}次") def avoid_pedestrian_in_crosswalk_detector(self): crosswalk_simTime = self.ego_data[self.ego_data["crossid"] != 20000][ "simTime" ].tolist() crosswalk_simTime_devide = self.different_road_area_simtime(crosswalk_simTime) for crosswalk_simtime in crosswalk_simTime_devide: if not self.pedestrian_data.empty: crosswalk_objstate = self.pedestrian_data[ self.pedestrian_data["simTime"].isin(crosswalk_simtime) ] else: crosswalk_objstate = pd.DataFrame() if len(crosswalk_objstate) > 0: pedestrian_simtime = crosswalk_objstate["simTime"] pedestrian_objstate = crosswalk_objstate[ crosswalk_objstate["simTime"].isin(pedestrian_simtime) ] ego_speed = np.sqrt( pedestrian_objstate["speedX"] ** 2 + pedestrian_objstate["speedY"] ** 2 ) if ego_speed.any() > 0: self.avoid_pedestrian_in_crosswalk_count += 1 def avoid_pedestrian_in_the_road_detector(self): simtime = self.pedestrian_in_front_of_car() if len(simtime) == 0: self.avoid_pedestrian_in_the_road_count += 0 else: pedestrian_on_the_road = self.pedestrian_data[ self.pedestrian_data["simTime"].isin(simtime) ] simTime = pedestrian_on_the_road["simTime"].tolist() simTime_devide = self.different_road_area_simtime(simTime) for simtime1 in simTime_devide: sub_pedestrian_on_the_road = pedestrian_on_the_road[ pedestrian_on_the_road["simTime"].isin(simtime1) ] ego_car = self.ego_data.loc[(self.ego_data["simTime"].isin(simtime1))] dist = np.sqrt( (ego_car["posX"].values - sub_pedestrian_on_the_road["posX"].values) ** 2 + ( ego_car["posY"].values - sub_pedestrian_on_the_road["posY"].values ) ** 2 ) speed = np.sqrt( ego_car["speedX"].values ** 2 + ego_car["speedY"].values ** 2 ) data = {"dist": dist, "speed": speed} new_ego_car = pd.DataFrame(data) new_ego_car = new_ego_car.assign( Column3=lambda x: (x["dist"] < 1) & (x["speed"] == 0) ) if new_ego_car["Column3"].any(): self.avoid_pedestrian_in_the_road_count += 1 def aviod_pedestrian_when_turning_detector(self): pedestrian_simtime_list = self.pedestrian_in_front_of_car() if len(pedestrian_simtime_list) > 0: simtime_list = self.ego_data[ (self.ego_data["simTime"].isin(pedestrian_simtime_list)) & (self.ego_data["lane_type"] == 20) ]["simTime"].tolist() simTime_list = self.different_road_area_simtime(simtime_list) pedestrian_on_the_road = self.pedestrian_data[ self.pedestrian_data["simTime"].isin(simtime_list) ] for simtime in simTime_list: sub_pedestrian_on_the_road = pedestrian_on_the_road[ pedestrian_on_the_road["simTime"].isin(simtime) ] ego_car = self.ego_data.loc[(self.ego_data["simTime"].isin(simtime))] ego_car["dist"] = np.sqrt( (ego_car["posX"].values - sub_pedestrian_on_the_road["posX"].values) ** 2 + ( ego_car["posY"].values - sub_pedestrian_on_the_road["posY"].values ) ** 2 ) ego_car["speed"] = np.sqrt( ego_car["speedX"].values ** 2 + ego_car["speedY"].values ** 2 ) if any(ego_car["speed"].tolist()) != 0: self.aviod_pedestrian_when_turning_count += 1 def calculate_slow_down_in_crosswalk_count(self): self.slow_down_in_crosswalk_detector() return self.slow_down_in_crosswalk_count def calculate_avoid_pedestrian_in_the_crosswalk_count(self): self.avoid_pedestrian_in_crosswalk_detector() return self.avoid_pedestrian_in_crosswalk_count def calculate_avoid_pedestrian_in_the_road_count(self): self.avoid_pedestrian_in_the_road_detector() return self.avoid_pedestrian_in_the_road_count def calculate_avoid_pedestrian_when_turning_count(self): self.aviod_pedestrian_when_turning_detector() return self.aviod_pedestrian_when_turning_count class TurnaroundViolation(object): def __init__(self, df_data): print("掉头违规类初始化中...") self.traffic_violations_type = "掉头违规类" # 存储原始数据引用 self._raw_data = df_data.obj_data[1] self.object_items = set(df_data.object_df.type.tolist()) # 存储行人数据引用 self._pedestrian_df = None if 13 in self.object_items: # 行人的type是13 self._pedestrian_df = df_data.object_df[df_data.object_df.type == 13] # 初始化属性，但不立即创建数据副本 self._ego_data = None self._pedestrian_data = None # 初始化计数器 self.turning_in_forbiden_turn_back_sign_count = 0 self.turning_in_forbiden_turn_left_sign_count = 0 self.avoid_pedestrian_when_turn_back_count = 0 @property def ego_data(self): """懒加载方式获取ego数据""" if self._ego_data is None: self._ego_data = self._raw_data[TURNAROUND_INFO].copy().reset_index(drop=True) return self._ego_data @property def pedestrian_data(self): """懒加载方式获取行人数据""" if self._pedestrian_data is None: if self._pedestrian_df is not None: self._pedestrian_data = self._pedestrian_df[SLOWDOWN_INFO].copy().reset_index(drop=True) else: self._pedestrian_data = pd.DataFrame() return self._pedestrian_data def pedestrian_in_front_of_car(self): if len(self.pedestrian_data) == 0: return [] else: self.ego_data["dx"] = self.ego_data["posX"] - self.pedestrian_data["posX"] self.ego_data["dy"] = self.ego_data["posY"] - self.pedestrian_data["posY"] self.ego_data["dist"] = np.sqrt( self.ego_data["dx"] ** 2 + self.ego_data["dy"] ** 2 ) self.ego_data["rela_pos"] = ( self.ego_data["dx"] * self.ego_data["speedX"] + self.ego_data["dy"] * self.ego_data["speedY"] ) simtime = self.ego_data[ (self.ego_data["rela_pos"] > 0) & (self.ego_data["dist"] < 50) ]["simTime"].tolist() return simtime def different_road_area_simtime(self, df, threshold=0.5): if not df: return [] simtime_group = [] current_simtime_group = [df[0]] for i in range(1, len(df)): if abs(df[i] - df[i - 1]) <= threshold: current_simtime_group.append(df[i]) else: simtime_group.append(current_simtime_group) current_simtime_group = [df[i]] simtime_group.append(current_simtime_group) return simtime_group def turn_back_in_forbiden_sign_detector(self): """ 禁止掉头type = 8 """ forbiden_turn_back_simTime = self.ego_data[self.ego_data["sign_type1"] == 8][ "simTime" ].tolist() forbiden_turn_left_simTime = self.ego_data[self.ego_data["sign_type1"] == 9][ "simTime" ].tolist() forbiden_turn_back_simtime_devide = self.different_road_area_simtime( forbiden_turn_back_simTime ) forbiden_turn_left_simtime_devide = self.different_road_area_simtime( forbiden_turn_left_simTime ) for forbiden_turn_back_simtime in forbiden_turn_back_simtime_devide: ego_car1 = self.ego_data.loc[ (self.ego_data["simFrame"].isin(forbiden_turn_back_simtime)) ] ego_start_speedx1 = ego_car1["speedX"].iloc[0] ego_start_speedy1 = ego_car1["speedY"].iloc[0] ego_end_speedx1 = ego_car1["speedX"].iloc[-1] ego_end_speedy1 = ego_car1["speedY"].iloc[-1] if ( ego_end_speedx1 * ego_start_speedx1 + ego_end_speedy1 * ego_start_speedy1 < 0 ): self.turning_in_forbiden_turn_back_sign_count += 1 for forbiden_turn_left_simtime in forbiden_turn_left_simtime_devide: ego_car2 = self.ego_data.loc[ (self.ego_data["simFrame"].isin(forbiden_turn_left_simtime)) ] ego_start_speedx2 = ego_car2["speedX"].iloc[0] ego_start_speedy2 = ego_car2["speedY"].iloc[0] ego_end_speedx2 = ego_car2["speedX"].iloc[-1] ego_end_speedy2 = ego_car2["speedY"].iloc[-1] if ( ego_end_speedx2 * ego_start_speedx2 + ego_end_speedy2 * ego_start_speedy2 < 0 ): self.turning_in_forbiden_turn_left_sign_count += 1 def avoid_pedestrian_when_turn_back_detector(self): sensor_on_intersection = self.pedestrian_in_front_of_car() avoid_pedestrian_when_turn_back_simTime_list = self.ego_data[ self.ego_data["lane_type"] == 20 ]["simTime"].tolist() avoid_pedestrian_when_turn_back_simTime_devide = ( self.different_road_area_simtime( avoid_pedestrian_when_turn_back_simTime_list ) ) if len(sensor_on_intersection) > 0: for avoid_pedestrian_when_turn_back_simtime in avoid_pedestrian_when_turn_back_simTime_devide: pedestrian_in_intersection_simtime = self.pedestrian_data[ self.pedestrian_data["simTime"].isin( avoid_pedestrian_when_turn_back_simtime ) ].tolist() ego_df = self.ego_data[ self.ego_data["simTime"].isin(pedestrian_in_intersection_simtime) ].reset_index(drop=True) pedestrian_df = self.pedestrian_data[ self.pedestrian_data["simTime"].isin( pedestrian_in_intersection_simtime ) ].reset_index(drop=True) ego_df["dist"] = np.sqrt( (ego_df["posx"] - pedestrian_df["posx"]) ** 2 + (ego_df["posy"] - pedestrian_df["posy"]) ** 2 ) ego_df["speed"] = np.sqrt(ego_df["speedx"] ** 2 + ego_df["speedy"] ** 2) if any(ego_df["speed"].tolist()) != 0: self.avoid_pedestrian_when_turn_back_count += 1 def calculate_turn_in_forbiden_turn_left_sign_count(self): self.turn_back_in_forbiden_sign_detector() return self.turning_in_forbiden_turn_left_sign_count def calculate_turn_in_forbiden_turn_back_sign_count(self): self.turn_back_in_forbiden_sign_detector() return self.turning_in_forbiden_turn_back_sign_count def calaulate_avoid_pedestrian_when_turn_back_count(self): self.avoid_pedestrian_when_turn_back_detector() return self.avoid_pedestrian_when_turn_back_count class WrongWayViolation(object): """停车违规类""" def __init__(self, df_data): print("停车违规类初始化中...") self.traffic_violations_type = "停车违规类" # 存储原始数据引用 self._raw_data = df_data.obj_data[1] # 初始化属性，但不立即创建数据副本 self._data = None # 初始化违规统计 self.violation_count = { "urbanExpresswayOrHighwayDrivingLaneStopped": 0, "urbanExpresswayOrHighwayEmergencyLaneStopped": 0, "urbanExpresswayEmergencyLaneDriving": 0, } @property def data(self): """懒加载方式获取数据""" if self._data is None: # 使用浅拷贝代替深拷贝 self._data = self._raw_data.copy() return self._data def process_violations(self): """处理停车或者紧急车道行驶违规数据""" # 提取有效道路类型 urban_expressway_or_highway = {1, 2} driving_lane = {1, 4, 5, 6} emergency_lane = {12} self.data["v"] *= 3.6 # 转换速度 # 使用向量化和条件判断进行违规判定 conditions = [ ( self.data["road_fc"].isin(urban_expressway_or_highway) & self.data["lane_type"].isin(driving_lane) & (self.data["v"] == 0) ), ( self.data["road_fc"].isin(urban_expressway_or_highway) & self.data["lane_type"].isin(emergency_lane) & (self.data["v"] == 0) ), ( self.data["road_fc"].isin(urban_expressway_or_highway) & self.data["lane_type"].isin(emergency_lane) & (self.data["v"] != 0) ), ] violation_types = [ "urbanExpresswayOrHighwayDrivingLaneStopped", "urbanExpresswayOrHighwayEmergencyLaneStopped", "urbanExpresswayEmergencyLaneDriving", ] # 设置违规类型 self.data["violation_type"] = None for condition, violation_type in zip(conditions, violation_types): self.data.loc[condition, "violation_type"] = violation_type # 统计违规情况 self.violation_count = ( self.data["violation_type"] .value_counts() .reindex(violation_types, fill_value=0) .to_dict() ) def calculate_urbanExpresswayOrHighwayDrivingLaneStopped_count(self): self.process_violations() return self.violation_count["urbanExpresswayOrHighwayDrivingLaneStopped"] def calculate_urbanExpresswayOrHighwayEmergencyLaneStopped_count(self): self.process_violations() return self.violation_count["urbanExpresswayEmergencyLaneDriving"] def calculate_urbanExpresswayEmergencyLaneDriving(self): self.process_violations() return self.violation_count["urbanExpresswayEmergencyLaneDriving"] class SpeedingViolation(object): """超速违规类""" """ 这里没有道路标志牌限速指标,因为shp地图中没有这个信息""" def __init__(self, df_data): print("超速违规类初始化中...") self.traffic_violations_type = "超速违规类" # 存储原始数据引用 self._raw_data = df_data.obj_data[1] # 初始化属性，但不立即创建数据副本 self._data = None # 初始化违规统计 self.violation_counts = { "urbanExpresswayOrHighwaySpeedOverLimit50": 0, "urbanExpresswayOrHighwaySpeedOverLimit20to50": 0, "urbanExpresswayOrHighwaySpeedOverLimit0to20": 0, "urbanExpresswayOrHighwaySpeedUnderLimit": 0, "generalRoadSpeedOverLimit50": 0, "generalRoadSpeedOverLimit20to50": 0, } @property def data(self): """懒加载方式获取数据""" if self._data is None: # 使用浅拷贝代替深拷贝 self._data = self._raw_data.copy() # 预处理数据 - 转换速度单位 self._data["v"] *= 3.6 # 转换为 km/h return self._data def process_violations(self): """处理数据帧，检查超速和其他违规行为""" # 提取有效道路类型 urban_expressway_or_highway = {1, 2} # 使用大括号直接创建集合 general_road = {3} # 直接创建包含一个元素的集合 self.data["v"] *= 3.6 # 转换速度 # 违规判定 conditions = [ ( self.data["road_fc"].isin(urban_expressway_or_highway) & (self.data["v"] > self.data["road_speed_max"] * 1.5) ), ( self.data["road_fc"].isin(urban_expressway_or_highway) & (self.data["v"] > self.data["road_speed_max"] * 1.2) & (self.data["v"] <= self.data["road_speed_max"] * 1.5) ), ( self.data["road_fc"].isin(urban_expressway_or_highway) & (self.data["v"] > self.data["road_speed_max"]) & (self.data["v"] <= self.data["road_speed_max"] * 1.2) ), ( self.data["road_fc"].isin(urban_expressway_or_highway) & (self.data["v"] < self.data["road_speed_min"]) ), ( self.data["road_fc"].isin(general_road) & (self.data["v"] > self.data["road_speed_max"] * 1.5) ), ( self.data["road_fc"].isin(general_road) & (self.data["v"] > self.data["road_speed_max"] * 1.2) & (self.data["v"] <= self.data["road_speed_max"] * 1.5) ), ] violation_types = [ "urbanExpresswayOrHighwaySpeedOverLimit50", "urbanExpresswayOrHighwaySpeedOverLimit20to50", "urbanExpresswayOrHighwaySpeedOverLimit0to20", "urbanExpresswayOrHighwaySpeedUnderLimit", "generalRoadSpeedOverLimit50", "generalRoadSpeedOverLimit20to50", ] # 设置违规类型 self.data["violation_type"] = None for condition, violation_type in zip(conditions, violation_types): self.data.loc[condition, "violation_type"] = violation_type # 统计各类违规情况 self.violation_counts = self.data["violation_type"].value_counts().to_dict() def calculate_urbanExpresswayOrHighwaySpeedOverLimit50_count(self): self.process_violations() return self.violation_counts.get("urbanExpresswayOrHighwaySpeedOverLimit50") if self.violation_counts.get( "urbanExpresswayOrHighwaySpeedOverLimit50") else 0 def calculate_urbanExpresswayOrHighwaySpeedOverLimit20to50_count(self): self.process_violations() return self.violation_counts["urbanExpresswayOrHighwaySpeedOverLimit20to50"] if self.violation_counts.get( "urbanExpresswayOrHighwaySpeedOverLimit20to50") else 0 def calculate_urbanExpresswayOrHighwaySpeedOverLimit0to20_count(self): self.process_violations() return self.violation_counts["urbanExpresswayOrHighwaySpeedOverLimit0to20"] if self.violation_counts.get( "urbanExpresswayOrHighwaySpeedOverLimit0to20") else 0 def calculate_urbanExpresswayOrHighwaySpeedUnderLimit_count(self): self.process_violations() return self.violation_counts["urbanExpresswayOrHighwaySpeedUnderLimit"] if self.violation_counts.get( "urbanExpresswayOrHighwaySpeedUnderLimit") else 0 def calculate_generalRoadSpeedOverLimit50(self): self.process_violations() return self.violation_counts["generalRoadSpeedOverLimit50"] if self.violation_counts.get( "generalRoadSpeedOverLimit50") else 0 def calculate_generalRoadSpeedOverLimit20to50_count(self): self.process_violations() return self.violation_counts["generalRoadSpeedOverLimit20to50"] if self.violation_counts.get( "generalRoadSpeedOverLimit20to50") else 0 class TrafficLightViolation(object): """违反交通灯类""" """需要补充判断车辆是左转直行还是右转，判断红绿灯是方向性红绿灯还是通过性红绿灯""" def __init__(self, df_data): """初始化方法""" self.traffic_violations_type = "违反交通灯类" print("违反交通灯类 类初始化中...") self.config = df_data.vehicle_config self.data_ego = df_data.ego_data # 获取数据 self.violation_counts = { "trafficSignalViolation": 0, "illegalDrivingOrParkingAtCrossroads": 0, } # 处理数据并判定违规 self.process_violations() def is_point_cross_line(self, point, stop_line_points): """ 判断车辆的某一坐标点是否跨越了由两个点定义的停止线（线段）。 使用向量叉积判断点是否在线段上，并通过计算车辆的航向角来判断是否跨越了停止线。 :param point: 车辆位置点 (x, y, heading)，包括 x, y 位置以及朝向角度（弧度制） :param stop_line_points: 停止线两个端点 [[x1, y1], [x2, y2]] :return: True 如果车辆跨越了停止线，否则 False """ line_vector = np.array( [ stop_line_points[1][0] - stop_line_points[0][0], stop_line_points[1][1] - stop_line_points[0][1], ] ) point_vector = np.array( [point[0] - stop_line_points[0][0], point[1] - stop_line_points[0][1]] ) cross_product = np.cross(line_vector, point_vector) if cross_product != 0: return False mid_point = ( np.array([stop_line_points[0][0], stop_line_points[0][1]]) + 0.5 * line_vector ) axletree_to_mid_vector = np.array( [point[0] - mid_point[0], point[1] - mid_point[1]] ) direction_vector = np.array([math.cos(point[2]), math.sin(point[2])]) norm_axletree_to_mid = np.linalg.norm(axletree_to_mid_vector) norm_direction = np.linalg.norm(direction_vector) if norm_axletree_to_mid == 0 or norm_direction == 0: return False cos_theta = np.dot(axletree_to_mid_vector, direction_vector) / ( norm_axletree_to_mid * norm_direction ) angle_theta = math.degrees(math.acos(cos_theta)) return angle_theta <= 90 def _filter_data(self): """过滤数据，筛选出需要分析的记录""" return self.data_ego[ (self.data_ego["stopline_id"] != -1) & (self.data_ego["stopline_type"] == 1) & (self.data_ego["trafficlight_id"] != -1) ] def _group_data(self, filtered_data): """按时间差对数据进行分组""" filtered_data["time_diff"] = filtered_data["simTime"].diff().fillna(0) threshold = 0.5 filtered_data["group"] = (filtered_data["time_diff"] > threshold).cumsum() return filtered_data.groupby("group") def _analyze_group(self, group_data): """分析单个分组的数据，判断是否闯红灯""" photos = [] stop_in_intersection = False for _, row in group_data.iterrows(): vehicle_pos = np.array([row["posX"], row["posY"], row["posH"]]) stop_line_points = [ [row["stopline_x1"], row["stopline_y1"]], [row["stopline_x2"], row["stopline_y2"]], ] traffic_light_status = row["traffic_light_status"] heading_vector = np.array([np.cos(row["posH"]), np.sin(row["posH"])]) heading_vector = heading_vector / np.linalg.norm(heading_vector) # with open(self.config_path / "vehicle_config.yaml", 'r') as f: # config = yaml.load(f, Loader=yaml.FullLoader) front_wheel_pos = vehicle_pos[:2] + self.config["EGO_WHEELBASS"] * heading_vector rear_wheel_pos = vehicle_pos[:2] - self.config["EGO_WHEELBASS"] * heading_vector dist = math.sqrt( (row["posX"] - row["traffic_light_x"]) ** 2 + (row["posY"] - row["traffic_light_y"]) ** 2 ) if abs(row["speedH"]) > 0.01 or abs(row["speedH"]) < 0.01: has_crossed_line_front = ( self.is_point_cross_line(front_wheel_pos, stop_line_points) and traffic_light_status == 1 ) has_crossed_line_rear = ( self.is_point_cross_line(rear_wheel_pos, stop_line_points) and row["v"] > 0 and traffic_light_status == 1 ) has_stop_in_intersection = has_crossed_line_front and row["v"] == 0 has_passed_intersection = has_crossed_line_front and dist < 1.0 # print(f'time: {row["simTime"]}, speed: {row["speedH"]}, posH: {row["posH"]}, dist: {dist:.2f}, has_stop_in_intersection: {has_stop_in_intersection}, has_passed_intersection: {has_passed_intersection}') photos.extend( [ has_crossed_line_front, has_crossed_line_rear, has_passed_intersection, has_stop_in_intersection, ] ) stop_in_intersection = has_passed_intersection return photos, stop_in_intersection def is_vehicle_run_a_red_light(self): """判断车辆是否闯红灯""" filtered_data = self._filter_data() grouped_data = self._group_data(filtered_data) self.photos_group = [] self.stop_in_intersections = [] for _, group_data in grouped_data: photos, stop_in_intersection = self._analyze_group(group_data) self.photos_group.append(photos) self.stop_in_intersections.append(stop_in_intersection) def process_violations(self): """处理数据并判定违规""" self.is_vehicle_run_a_red_light() count_1 = sum(all(photos) for photos in self.photos_group) count_2 = sum( stop_in_intersection for stop_in_intersection in self.stop_in_intersections ) self.violation_counts["trafficSignalViolation"] = count_1 self.violation_counts["illegalDrivingOrParkingAtCrossroads"] = count_2 def calculate_trafficSignalViolation_count(self): self.process_violations() return self.violation_counts["trafficSignalViolation"] def calculate_illegalDrivingOrParkingAtCrossroads(self): self.process_violations() return self.violation_counts["illegalDrivingOrParkingAtCrossroads"] class WarningViolation(object): """警告性违规类""" def __init__(self, df_data): print("警告性违规类初始化中...") self.traffic_violations_type = "警告性违规类" # 存储原始数据引用 self.config = df_data.vehicle_config self._raw_data = df_data.obj_data[1] # 初始化属性，但不立即创建数据副本 self._data = None # 初始化违规计数器 self.violation_counts = { "generalRoadIrregularLaneUse": 0, # 驾驶机动车在高速公路、城市快速路以外的道路上不按规定车道行驶 "urbanExpresswayOrHighwayRideLaneDivider": 0, # 机动车在高速公路或者城市快速路上骑、轧车行道分界线 } @property def data(self): """懒加载方式获取数据""" if self._data is None: # 使用浅拷贝代替深拷贝 self._data = self._raw_data.copy() return self._data def process_violations(self): """处理所有违规类型""" # 处理普通道路不按规定车道行驶违规 self._process_irregular_lane_use() # 处理骑、轧车行道分界线违规 self._process_lane_divider_violation() def _process_irregular_lane_use(self): """处理普通道路不按规定车道行驶违规""" # 定义道路和车道类型 general_road = {3} # 普通道路 lane_type = {11} # 非机动车道 # 使用布尔索引来筛选满足条件的行 condition = (self.data["road_fc"].isin(general_road)) & ( self.data["lane_type"].isin(lane_type) ) # 创建一个新的列，并根据条件设置值 self.data["is_violation"] = condition # 统计满足条件的连续时间段 violation_segments = self.count_continuous_violations( self.data["is_violation"], self.data["simTime"] ) # 更新违规计数 self.violation_counts["generalRoadIrregularLaneUse"] = len(violation_segments) def _process_lane_divider_violation(self): """处理骑、轧车行道分界线违规""" # 获取车辆和车道宽度 car_width = self.config["CAR_WIDTH"] lane_width = self.data["lane_width"] # 计算阈值 threshold = (lane_width - car_width) / 2 # 找到满足条件的行 self.data["is_violation"] = self.data["laneOffset"] > threshold # 统计满足条件的连续时间段 violation_segments = self.count_continuous_violations( self.data["is_violation"], self.data["simTime"] ) # 更新违规计数 self.violation_counts["urbanExpresswayOrHighwayRideLaneDivider"] = len( violation_segments ) def count_continuous_violations(self, violation_series, time_series): """统计连续违规的时间段数量 Args: violation_series: 表示是否违规的布尔序列 time_series: 对应的时间序列 Returns: list: 连续违规时间段列表 """ continuous_segments = [] current_segment = [] for is_violation, time in zip(violation_series, time_series): if is_violation: if not current_segment: # 新的连续段开始 current_segment.append(time) else: if current_segment: # 连续段结束 current_segment.append(time) # 添加结束时间 continuous_segments.append(current_segment) current_segment = [] # 检查是否有一个未结束的连续段在最后 if current_segment: current_segment.append(time_series.iloc[-1]) # 使用最后的时间作为结束时间 continuous_segments.append(current_segment) return continuous_segments def calculate_generalRoadIrregularLaneUse_count(self): """计算普通道路不按规定车道行驶违规次数""" # 只处理普通道路不按规定车道行驶违规 self._process_irregular_lane_use() return self.violation_counts["generalRoadIrregularLaneUse"] def calculate_urbanExpresswayOrHighwayRideLaneDivider_count(self): """计算骑、轧车行道分界线违规次数""" # 只处理骑、轧车行道分界线违规 self._process_lane_divider_violation() return self.violation_counts["urbanExpresswayOrHighwayRideLaneDivider"] class TrafficSignViolation: """交通标志违规类""" PROHIBITED_STRAIGHT_THRESHOLD = 5 SIGN_TYPE_STRAIGHT_PROHIBITED = 7 SIGN_TYPE_SPEED_LIMIT = 12 SIGN_TYPE_MIN_SPEED_LIMIT = 13 def __init__(self, df_data): print("交通标志违规类初始化中...") self.traffic_violations_type = "交通标志违规类" # 存储原始数据引用 self._raw_data = df_data.obj_data[1] # 初始化属性，但不立即创建数据副本 self._data = None # 延迟计算标志 self._calculated = False self._violation_counts = { "NoStraightThrough": 0, "SpeedLimitViolation": 0, "MinimumSpeedLimitViolation": 0 } @property def data(self): """懒加载方式获取数据""" if self._data is None: # 使用浅拷贝代替深拷贝 self._data = self._raw_data.copy() # 预处理数据 - 按时间排序 self._data = self._data.sort_values('simTime').reset_index(drop=True) return self._data def _ensure_calculated(self): """保证计算只执行一次""" if not self._calculated: self._check_prohibition_violations() self._check_instruction_violations() self._calculated = True def calculate_NoStraightThrough_count(self): """计算禁止直行违规次数""" self._ensure_calculated() return self._violation_counts["NoStraightThrough"] def calculate_SpeedLimitViolation_count(self): """计算超速违规次数""" self._ensure_calculated() return self._violation_counts["SpeedLimitViolation"] def calculate_MinimumSpeedLimitViolation_count(self): """计算最低限速违规次数""" self._ensure_calculated() return self._violation_counts["MinimumSpeedLimitViolation"] def _check_prohibition_violations(self): """处理禁令标志违规（禁止直行和限速）""" self._check_straight_violation() self._check_speed_violation( self.SIGN_TYPE_SPEED_LIMIT, operator.gt, "SpeedLimitViolation" ) def _check_instruction_violations(self): """处理指示标志违规（最低限速）""" self._check_speed_violation( self.SIGN_TYPE_MIN_SPEED_LIMIT, operator.lt, "MinimumSpeedLimitViolation" ) def _check_straight_violation(self): """检查禁止直行违规""" straight_df = self.data[self.data["sign_type1"] == self.SIGN_TYPE_STRAIGHT_PROHIBITED] if not straight_df.empty: # 计算航向角变化并填充缺失值 straight_df = straight_df.copy() straight_df['posH_diff'] = straight_df['posH'].diff().abs().fillna(0) # 创建筛选条件 mask = ( (straight_df['posH_diff'] <= self.PROHIBITED_STRAIGHT_THRESHOLD) & (straight_df['v'] > 0) ) self._violation_counts["NoStraightThrough"] = mask.sum() def _check_speed_violation(self, sign_type, compare_op, count_key): """通用速度违规检查方法 Args: sign_type: 标志类型 compare_op: 比较操作符 count_key: 违规计数键名 """ violation_df = self.data[self.data["sign_type1"] == sign_type] if not violation_df.empty: mask = compare_op(violation_df['v'], violation_df['sign_speed']) self._violation_counts[count_key] = mask.sum()