#!/usr/bin/env python # -*- coding: utf-8 -*- ################################################################## # # Copyright (c) 2023 CICV, Inc. All Rights Reserved # ################################################################## """ @Authors: xieguijin(xieguijin@china-icv.cn), yangzihao(yangzihao@china-icv.cn) @Data: 2023/08/21 @Last Modified: 2023/08/21 @Summary: Compliance metrics """ import sys sys.path.append('../common') sys.path.append('../modules') sys.path.append('../results') import numpy as np import pandas as pd from common import score_grade, string_concatenate, replace_key_with_value, score_over_100 from scipy.spatial.distance import euclidean class traffic_rule(object): """ Class for achieving compliance metrics for autonomous driving. Attributes: droadMark_df: Roadmark data, stored in dataframe format. """ def __init__(self, data_processed, scoreModel): self.scoreModel = scoreModel self.roadMark_df = data_processed.road_mark_df self.trafficLight_df = data_processed.traffic_light_df self.trafficSignal_df = data_processed.traffic_signal_df self.objState_df = data_processed.object_df self.ego_df = self.objState_df[(self.objState_df.playerId == 1) & (self.objState_df.type == 1)] self.violation_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'violation']) def _filter_groups_by_frame_period(self, grouped_violations): """ Filter groups by a minimum continuous frame period. """ CONTINUOUS_FRAME_PERIOD = 13 return [g for g in grouped_violations if len(g[0]) >= CONTINUOUS_FRAME_PERIOD] def _extract_violation_times(self, filtered_groups): """ Create a dataframe with start and end times for each violation group. """ return [[g[0][0], g[0][-1]] for g in filtered_groups] def _filter_solid_lines(self, Dimy): """ Filter solid lines within the player's lateral distance. """ dist_line = self.roadMark_df[self.roadMark_df["type"] == 1] dist_line = dist_line.reset_index() return dist_line[abs(dist_line["lateralDist"].values) <= Dimy] def _group_violations(self, dist_press): """ Group violations by continuous frames. """ t_list = dist_press['simTime'].values.tolist() f_list = dist_press['simFrame'].values.tolist() group_time = [] group_frame = [] sub_group_time = [] sub_group_frame = [] for i in range(len(f_list)): if not sub_group_time or t_list[i] - t_list[i - 1] <= 1: sub_group_time.append(t_list[i]) sub_group_frame.append(f_list[i]) else: group_time.append(sub_group_time) group_frame.append(sub_group_frame) sub_group_time = [t_list[i]] sub_group_frame = [f_list[i]] group_time.append(sub_group_time) group_frame.append(sub_group_frame) return list(zip(group_time, group_frame)) def get_solid_line_violations(self): """ Detect violations of pressing solid lines and return a dictionary with violation details. """ # Extract relevant data Dimy = self.objState_df[self.objState_df["playerId"] == 1]["dimY"][0] / 2 dist_press = self._filter_solid_lines(Dimy) grouped_violations = self._group_violations(dist_press)# Group violations by continuous frames filtered_groups = self._filter_groups_by_frame_period(grouped_violations)# Filter groups by minimum frame period # Calculate violation count and create violation dataframe press_line_count = len(filtered_groups) press_line_time = self._extract_violation_times(filtered_groups) if press_line_time: time_df = pd.DataFrame(press_line_time, columns=['start_time', 'end_time']) time_df['violation'] = '压实线' # Update violation dataframe self.violation_df = pd.concat([self.violation_df, press_line_time], ignore_index=True) # Create and return violation dictionary warning_count = 0 press_line_dict = { 'metric': 'pressSolidLine', 'weight': 3, 'illegal_count': press_line_count, 'penalty_points': press_line_count * 3, 'penalty_money': press_line_count * 200, 'warning_count': warning_count, 'penalty_law': '《中华人民共和国道路交通安全法》第八十二条:机动车在高速公路上行驶，不得有下列行为:(三)骑、轧车行道分界线或者在路肩上行驶。' } return press_line_dict def normalize_angle(self, angle): """Normalize angle to the range [0, 360).""" difference = angle while difference >= 360: difference -= 360 return difference def is_red_light(self, simTime, cycleTime, duration_start, duration_end): """Check if the current time corresponds to a red light phase.""" divisor = simTime / cycleTime decimal_part = divisor - int(divisor) return duration_start <= decimal_part < duration_end def process_traffic_light(self, trafficLight_id): """Process a single traffic light and detect run red light events.""" trafficLight_position = self.trafficSignal_df[self.trafficSignal_df["playerId"] == trafficLight_id].iloc[:1, :] if trafficLight_position.empty: return trafficLight_position_x = trafficLight_position['posX'].values[0] trafficLight_position_y = trafficLight_position['posY'].values[0] trafficLight_position_heading = trafficLight_position['posH'].values[0] trafficLight_character = self.trafficLight_df[self.trafficLight_df.id == trafficLight_id] cycleTime = trafficLight_character["cycleTime"].values[0] noPhases = trafficLight_character["noPhases"].values[0] # Calculate distances and headings self.ego_df["traffic_light_distance_absolute"] = self.ego_df[['posX', 'posY']].apply( lambda x: euclidean((trafficLight_position_x, trafficLight_position_y), (x['posX'], x['posY'])), axis=1) self.ego_df["traffic_light_h_diff"] = self.ego_df.apply( lambda x: abs(x['posH'] - trafficLight_position_heading) * 57.3, axis=1) self.ego_df["traffic_light_h_diff"] = self.ego_df["traffic_light_h_diff"].apply(self.normalize_angle) # Filter ego vehicles near the traffic light with the correct heading mask_traffic_light = ((self.ego_df['traffic_light_h_diff'] <= 210) & ( self.ego_df['traffic_light_h_diff'] >= 150)) | (self.ego_df['traffic_light_h_diff'] <= 30) | ( self.ego_df['traffic_light_h_diff'] >= 330) ego_near_light = self.ego_df[(self.ego_df.traffic_light_distance_absolute <= 10) & mask_traffic_light] if ego_near_light.empty: return # Check for red light violations ego_near_light["flag_red_traffic_light"] = 0 type_list = trafficLight_character['violation'][:noPhases] duration = trafficLight_character['duration'][:noPhases] duration_correct = [0] * noPhases for number in range(noPhases): duration_correct[number] = sum(duration[:number + 1]) type_current = type_list.values[number] if type_current == 1: # Red light phase if number == 0: duration_start = 0 else: duration_start = duration_correct[number - 1] duration_end = duration_correct[number] ego_near_light["flag_red_traffic_light"] = ego_near_light.apply( lambda x: self.is_red_light(x['simTime'], cycleTime, duration_start, duration_end), axis=1) # Collect run red light events run_red_light_df = ego_near_light[ego_near_light['flag_red_traffic_light'] == 1] self.collect_run_red_light_events(run_red_light_df) def collect_run_red_light_events(self, run_red_light_df): grouped_events = self._group_violations(run_red_light_df) filtered_events = self._filter_groups_by_frame_period(grouped_events) violation_times = self._extract_violation_times(filtered_events) if violation_times: time_df = pd.DataFrame(violation_times, columns=['start_time', 'end_time']) time_df['violation'] = '闯红灯' self.violation_df = pd.concat([self.violation_df, time_df], ignore_index=True) def run_red_light_detection(self): """Main function to detect run red light events.""" trafficLight_id_list = set(self.trafficLight_df["id"].tolist()) run_red_light_count = 0 for trafficLight_id in trafficLight_id_list: self.process_traffic_light(trafficLight_id) # 闯红灯次数统计（这里可以根据需要修改统计逻辑） if 'flag_red_traffic_light' in self.ego_df.columns and self.ego_df['flag_red_traffic_light'].any() == 1: run_red_light_count += 1 run_red_light_dict = { 'metric': 'runRedLight', 'weight': 6, 'illegal_count': run_red_light_count, 'penalty_points': run_red_light_count * 6, 'penalty_money': run_red_light_count * 200, 'warning_count': 0, 'penalty_law': '《中华人民共和国道路交通安全法实施条例》第四十条:(二)红色叉形灯或者箭头灯亮时，禁止本车道车辆通行。' } return run_red_light_dict def _find_speed_violations(self): DimX = self.objState_df[self.objState_df["playerId"] == 1]["dimY"][0] / 2 data_ego = self.objState_df[self.objState_df["playerId"] == 1] speed_limit_sign = self.trafficSignal_df[self.trafficSignal_df["type"] == 274] same_df_rate = pd.merge(speed_limit_sign, data_ego, on=['simTime', 'simFrame'], how='inner').reset_index() speed_df = same_df_rate[(abs(same_df_rate["posX_x"] - same_df_rate["posX_y"]) <= 7) & ( abs(same_df_rate["posY_x"] - same_df_rate["posY_y"]) <= DimX)] speed_df["speed"] = np.sqrt(speed_df["speedX"] ** 2 + speed_df["speedY"] ** 2) * 3.6 list_sign = speed_df[speed_df["speed"] > speed_df["value"]] return list_sign, speed_df def _calculate_overspeed_statistics(self, speed_df, list_sign): index_sign = list_sign.index.to_list() speed_df["flag_press"] = speed_df["simFrame"].apply(lambda x: 1 if x in list_sign["simFrame"] else 0) speed_df["diff_press"] = speed_df["flag_press"].diff() index_list = [] subindex_list = [] for i in range(len(index_sign)): if not subindex_list or index_sign[i] - index_sign[i - 1] == 1: subindex_list.append(index_sign[i]) else: index_list.append(subindex_list) subindex_list = [index_sign[i]] index_list.append(subindex_list) overspeed_count_0_to_10 = 0 overspeed_count_10_to_20 = 0 overspeed_count_20_to_50 = 0 overspeed_count_50_to_ = 0 if index_list[0]: for i in range(len(index_list)): left = index_list[i][0] right = index_list[i][-1] df_tmp = speed_df.loc[left:right + 1] max_ratio = ((df_tmp["speed"] - df_tmp["value"]) / df_tmp["value"]).max() if 0 <= max_ratio < 0.1: overspeed_count_0_to_10 += 1 elif 0.1 <= max_ratio < 0.2: overspeed_count_10_to_20 += 1 elif 0.2 <= max_ratio < 0.5: overspeed_count_20_to_50 += 1 elif max_ratio >= 0.5: overspeed_count_50_to_ += 1 return ( self._create_overspeed_dict(overspeed_count_0_to_10, 'overspeed10', 0, 0), self._create_overspeed_dict(overspeed_count_10_to_20, 'overspeed10_20', 0, 200), self._create_overspeed_dict(overspeed_count_20_to_50, 'overspeed20_50', 6, 200), self._create_overspeed_dict(overspeed_count_50_to_, 'overspeed50', 12, 2000) ) def _create_overspeed_dict(self, count, metric, penalty_points, penalty_money): return { 'metric': metric, 'weight': None, 'illegal_count': count, 'penalty_points': count * penalty_points, 'penalty_money': count * penalty_money, 'warning_count': count if penalty_points == 0 else 0, 'penalty_law': '《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。' } def overspeed(self): list_sign, speed_df = self._find_speed_violations() grouped_events = self._group_violations(list_sign) filtered_events = self._filter_groups_by_frame_period(grouped_events) violation_times = self._extract_violation_times(filtered_events) if violation_times: time_df = pd.DataFrame([d[0] for d in violation_times], columns=['start_time', 'end_time']) time_df['violation'] = '超速' self.violation_df = pd.concat([self.violation_df, time_df], ignore_index=True) return self._calculate_overspeed_statistics(speed_df, list_sign) class Compliance(object): def __init__(self, data_processed, custom_data, scoreModel): self.eval_data = pd.DataFrame() self.penalty_points = 0 self.config = data_processed.config compliance_config = self.config.config['compliance'] self.compliance_config = compliance_config self.weight_dict = compliance_config['weight'] self.metric_list = compliance_config['metric'] self.type_list = compliance_config['type'] print("self.type_list is", self.type_list) self.weight_custom = compliance_config['weightCustom'] self.name_dict = compliance_config['name'] self.metric_dict = compliance_config['typeMetricDict'] self.type_name_dict = compliance_config['typeName'] self.weight = compliance_config['weightDimension'] self.weight_type_dict = compliance_config['typeWeight'] self.weight_type_list = compliance_config['typeWeightList'] self.type_illegal_count_dict = {} self.traffic_rule = traffic_rule(data_processed, scoreModel) self.violation_df = self.traffic_rule.violation_df def score_cal_penalty_points(self, penalty_points): if penalty_points == 0: score = 100 elif penalty_points >= 12: score = 0 else: score = (12 - penalty_points) / 12 * 60 return score def time_splice(self, start_time, end_time): str_time = f"[{start_time}s, {end_time}s]" return str_time def weight_type_cal(self): # penalty_list = [1, 3, 6, 9, 12] penalty_list = [1, 3, 6, 12] sum_penalty = sum(penalty_list) weight_type_list = [round(x / sum_penalty, 2) for x in penalty_list] return weight_type_list def compliance_statistic(self): # metric analysis press_line_dict = self.traffic_rule.get_solid_line_violations() run_red_light_dict = self.traffic_rule.run_red_light_detection() overspeed_0_to_10_dict, overspeed_10_to_20_dict, overspeed_20_to_50_dict, overspeed_50_dict = self.traffic_rule.overspeed() df_list = [] if "overspeed10" in self.metric_list: df_list.append(overspeed_0_to_10_dict) if "overspeed10_20" in self.metric_list: df_list.append(overspeed_10_to_20_dict) if "pressSolidLine" in self.metric_list: df_list.append(press_line_dict) if "runRedLight" in self.metric_list: df_list.append(run_red_light_dict) if "overspeed20_50" in self.metric_list: df_list.append(overspeed_20_to_50_dict) if "overspeed50" in self.metric_list: df_list.append(overspeed_50_dict) # generate dataframe and dicts compliance_df = pd.DataFrame(df_list) return compliance_df def prepare_data(self): self.compliance_df = self.compliance_statistic() self.illegal_count = int(self.compliance_df['illegal_count'].sum()) self.metric_penalty_points_dict = self.compliance_df.set_index('metric').to_dict()['penalty_points'] self.metric_illegal_count_dict = self.compliance_df.set_index('metric').to_dict()['illegal_count'] self.metric_penalty_money_dict = self.compliance_df.set_index('metric').to_dict()['penalty_money'] self.metric_warning_count_dict = self.compliance_df.set_index('metric').to_dict()['warning_count'] self.metric_penalty_law_dict = self.compliance_df.set_index('metric').to_dict()['penalty_law'] # 初始化数据字典 self.illegal_count = int(self.compliance_df['illegal_count'].sum()) self.metric_penalty_points_dict = self.compliance_df.set_index('metric').to_dict()['penalty_points'] self.metric_illegal_count_dict = self.compliance_df.set_index('metric').to_dict()['illegal_count'] self.metric_penalty_money_dict = self.compliance_df.set_index('metric').to_dict()['penalty_money'] self.metric_warning_count_dict = self.compliance_df.set_index('metric').to_dict()['warning_count'] self.metric_penalty_law_dict = self.compliance_df.set_index('metric').to_dict()['penalty_law'] def calculate_deduct_scores(self): score_type_dict = {} deduct_functions = { "deduct1": self.calculate_deduct(1), "deduct3": self.calculate_deduct(3), "deduct6": self.calculate_deduct(6), "deduct9": self.calculate_deduct(9), "deduct12": self.calculate_deduct(12), } for deduct_type in self.type_list: print("deduct_type is", deduct_type) if deduct_type in deduct_functions: penalty_points, illegal_count = deduct_functions[deduct_type] score_type_dict[deduct_type] = self.score_cal_penalty_points(penalty_points) self.type_illegal_count_dict[deduct_type] = illegal_count return score_type_dict def calculate_deduct(self, num): deduct_df = self.compliance_df[(self.compliance_df['weight'].isna()) | (self.compliance_df['weight'] == num)] return deduct_df['penalty_points'].sum(), deduct_df['illegal_count'].sum() def calculate_weights(self): weight_dict = { "overspeed10": 0.5, "overspeed10_20": 0.5, "pressSolidLine": 1.0, "runRedLight": 0.5, "overspeed20_50": 0.5, "overspeed50": 1.0 } self.weight_type_list = [weight_dict.get(metric, 0.5) for metric in self.type_list] # 假设 type_list 中的每个元素都在 weight_dict 的键中，或者默认为 0.5 self.weight_type_dict = {key: value for key, value in zip(self.type_list, self.weight_type_list)} self.weight_dict = weight_dict def calculate_compliance_score(self, score_type_dict): if not self.weight_custom: # 客观赋权 self.calculate_weights() penalty_points_threshold = 12 # 假设的扣分阈值 if hasattr(self, 'penalty_points') and self.penalty_points >= penalty_points_threshold: score_compliance = 0 elif sum(score_type_dict.values()) / len(score_type_dict) == 100: score_compliance = 100 else: score_type_tmp = [80 if x == 100 else x for key, x in score_type_dict.items()] score_compliance = np.dot(self.weight_type_list, score_type_tmp) return round(score_compliance, 2) def output_results(self, score_compliance, score_type_dict): print("\n[合规性表现及得分情况]") print(f"合规性得分为:{score_compliance:.2f}分。") print(f"合规性各分组得分为:{score_type_dict}。") print(f"合规性各分组权重为:{self.weight_type_list}。") def compliance_score(self): self.prepare_data() score_type_dict = self.calculate_deduct_scores() score_compliance = self.calculate_compliance_score(score_type_dict) self.output_results(score_compliance, score_type_dict) return score_compliance, score_type_dict def _get_weight_distribution(self): # get weight distribution weight_distribution = {} weight_distribution["name"] = self.config.dimension_name["compliance"] for type in self.type_list: type_weight_indexes_dict = {key: f"{self.name_dict[key]}({value * 100:.2f}%)" for key, value in self.weight_dict.items() if key in self.metric_dict[type]} weight_distribution_type = { "weight": f"{self.type_name_dict[type]}({self.weight_type_dict[type] * 100:.2f}%)", "indexes": type_weight_indexes_dict } weight_distribution[type] = weight_distribution_type return weight_distribution def build_weight_index(self, metric_prefix, weight_key): if metric_prefix in self.metric_list: return { f"{metric_prefix}Weight": f"{metric_prefix.replace('_', ' ').capitalize()}({self.weight_dict[weight_key] * 100:.2f}%)"} return {} def compliance_weight_distribution(self): weight_distribution = {"name": "合规性"} deduct_types = { "deduct1": {"metrics": ["overspeed10", "overspeed10_20"], "weight_key": "deduct1"}, "deduct3": {"metrics": ["pressSolidLine"], "weight_key": "deduct3"}, "deduct6": {"metrics": ["runRedLight", "overspeed20_50"], "weight_key": "deduct6"}, "deduct9": {"metrics": [], "weight_key": "deduct9"}, # Assuming no specific metrics for deduct9 "deduct12": {"metrics": ["overspeed50"], "weight_key": "deduct12"}, } for deduct_type, details in deduct_types.items(): if deduct_type in self.type_list: indexes_dict = {} for metric in details["metrics"]: indexes_dict.update(self.build_weight_index(metric, details["weight_key"])) weight_distribution[deduct_type] = { f"{deduct_type.replace('deduct', '')}Weight": f"{details['weight_key'].replace('deduct', '').replace('_', ' ').capitalize()}" f"违规（{int(details['weight_key'].replace('deduct', ''))}分）({self.weight_type_dict[deduct_type] * 100:.2f} % )", "indexes": indexes_dict } if deduct_type == "deduct9" and "deduct9" in self.type_list: weight_distribution[deduct_type] = { "deduct9Weight": f"严重违规（9分）({self.weight_type_dict[deduct_type] * 100:.2f}%)", "indexes": {} } return weight_distribution def report_statistic(self): score_compliance, score_type_dict = self.compliance_score() grade_compliance = score_grade(score_compliance) score_compliance = int(score_compliance) if int(score_compliance) == score_compliance else score_compliance score_type = [int(n) if int(n) == n else n for key, n in score_type_dict.items()] # 获取合规性描述 comp_description1 = self.get_compliance_description(grade_compliance, self.illegal_count) comp_description2 = self.get_violation_details(self.type_list, self.type_illegal_count_dict, self.type_name_dict) weight_distribution = self._get_weight_distribution() # 获取违规数据表 violations_slices = self.get_violations_table() # 获取扣分详情 deductPoint_dict = self.get_deduct_points_dict(score_type_dict) # 返回结果（这里可以根据需要返回具体的数据结构） return { "name": "合规性", "weight": f"{self.weight * 100:.2f}%", "weightDistribution": weight_distribution, "score": score_compliance, "level": grade_compliance, 'score_type': score_type, # 'score_metric': score_metric, 'illegalCount': self.illegal_count, "description1": comp_description1, "description2": comp_description2, "details": deductPoint_dict, "violations": violations_slices } def get_compliance_description(self, grade_compliance, illegal_count): # 获取合规性描述 if grade_compliance == '优秀': return '车辆在本轮测试中无违反交通法规行为；' else: return f'车辆在本轮测试中共发生{illegal_count}次违反交通法规行为；' \ f'如果等级为一般或较差，需要提高算法在合规性上的表现。' def get_violation_details(self, type_list, type_illegal_count_dict, type_name_dict): # 获取违规详情描述 if self.illegal_count == 0: return "车辆在该用例中无违反交通法规行为，算法表现良好。" else: str_illegel_type = ", ".join( [f"{type_name_dict[type]}行为{count}次" for type, count in type_illegal_count_dict.items() if count > 0]) return f"车辆在该用例共违反交通法规{self.illegal_count}次。其中{str_illegel_type}。违规行为详情见附录C。" def get_violations_table(self): # 获取违规数据表 if not self.violation_df.empty: self.violation_df['time'] = self.violation_df.apply( lambda row: self.time_splice(row['start_time'], row['end_time']), axis=1) df_violations = self.violation_df[['time', 'violation']] return df_violations.to_dict('records') else: return [] def get_deduct_points_dict(self, score_type_dict): # 获取扣分详情字典 deductPoint_dict = {} for type in self.type_list: type_dict = { "name": self.type_name_dict[type], "score": score_type_dict[type], "indexes": {} } for metric in self.metric_list: type_dict["indexes"][metric] = { "name": self.name_dict[metric], "times": self.metric_illegal_count_dict[metric], "deductPoints": self.metric_penalty_points_dict[metric], "fine": self.metric_penalty_money_dict[metric], "basis": self.metric_penalty_law_dict[metric] } deductPoint_dict[type] = type_dict for deduct_type, metrics in [ ("deduct1", ["overspeed10", "overspeed10_20"]), ("deduct3", ["pressSolidLine"]), ("deduct6", ["runRedLight", "overspeed20_50"]), ("deduct9", ["xx"]), # 注意：这里xx是示例，实际应替换为具体的违规类型 ("deduct12", ["overspeed50"]) ]: if deduct_type in self.type_list: deduct_indexes = {} for metric in metrics: if metric in self.metric_list: deduct_indexes[metric] = { "name": self.name_dict[metric], "times": self.metric_illegal_count_dict[metric], "deductPoints": self.metric_penalty_points_dict[metric], "fine": self.metric_penalty_money_dict[metric], "basis": self.metric_penalty_law_dict[metric] } # 对于deduct9，特殊处理其score值 score = score_type_dict.get(deduct_type, 100) if deduct_type != "deduct9" else 100 deductPoint_dict[deduct_type] = { "name": f"{deduct_type.replace('deduct', '').strip()}违规（{int(deduct_type.replace('deduct', ''))}分）", "score": score, "indexes": deduct_indexes } return deductPoint_dict def get_eval_data(self): df = self.eval_data return df