bdzl2/ai-train_platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

道路灾害识别更换最新模型，道路信息excel模板，识别宽度提取模板调整

Browse Source
This commit is contained in:
liyubo 2026-02-27 16:42:44 +08:00
parent f349efaa56
commit 085e1f4044
3 changed files with 152 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
predict/predict_yolo11seg.py
View File

@ -95,7 +95,7 @@ class InferenceResult:
class YOLOSegmentationInference: class YOLOSegmentationInference:
"""YOLO分割模型推理工具""" """YOLO分割模型推理工具"""
def __init__(self, model_path: str, device: Optional[str] = None): def __init__(self, model_path: Optional[str] = None, device: Optional[str] = None):
""" """
初始化推理工具 初始化推理工具
@ -105,6 +105,7 @@ class YOLOSegmentationInference:
""" """
self.model_path = model_path self.model_path = model_path
self.device = device self.device = device
self.conf_threshold = 0.25
self.model = None self.model = None
self.class_names = [] self.class_names = []
self.clean_memory_per_count = 500 self.clean_memory_per_count = 500
@ -128,6 +129,11 @@ class YOLOSegmentationInference:
(64, 64, 64), # 浅灰色 (64, 64, 64), # 浅灰色
] ]
def set_model_info(self, model_path, device, conf_threshold) :
self.model_path = model_path
self.device = device
self.conf_threshold = conf_threshold
def load_model(self) -> bool: def load_model(self) -> bool:
""" """
加载YOLO分割模型 加载YOLO分割模型
@ -136,7 +142,7 @@ class YOLOSegmentationInference:
加载成功返回True，否则返回False 加载成功返回True，否则返回False
""" """
try: try:
print(f"正在加载模型: {self.model_path}") print(f"正在加载模型: {self.model_path} 置信度：{self.conf_threshold}")
self.model = YOLO(self.model_path) self.model = YOLO(self.model_path)
# 设置设备 # 设置设备
@ -639,6 +645,32 @@ class YOLOSegmentationInference:
print(f"处理目录失败: {e}") print(f"处理目录失败: {e}")
return results return results
def get_model_info(self, pile_dict, road_dict, image_path) :
tmp_image_path = smb_tool.standardized_path(image_path)
img_file_name = tmp_image_path.split('/')[-1]
road_data = get_road_info(road_dict, pile_dict, img_file_name)
print(f"road_data={road_data}")
# 获取当前脚本文件所在的目录
script_dir = os.path.dirname(__file__)
# 构建模型文件的绝对路径
asphalt_model_path = os.path.abspath(os.path.join(script_dir, '..', 'pt_save', 'asphalt20260212_0.391.pt'))
cream_model_path = os.path.abspath(os.path.join(script_dir, '..', 'pt_save', 'asphalt20260212_0.391.pt'))
if road_data :
road_type_cn = road_data['路面类型（沥青/水泥/砂石）']
if road_type_cn == '沥青' :
road_type = "asphalt"
return 0.391, asphalt_model_path
elif road_type_cn == '水泥' :
road_type = "cement"
return 0.1, cream_model_path
elif road_type_cn == '砾石' :
road_type = "gravel"
return 0.391, asphalt_model_path
return 0.391, asphalt_model_path
def process_image_directory_share_dir_circle(self, task_id, current_time, input_dir_list, user_name, pwd, output_dir: Optional[str] = None, def process_image_directory_share_dir_circle(self, task_id, current_time, input_dir_list, user_name, pwd, output_dir: Optional[str] = None,
conf_threshold: float = 0.25, iou_threshold: float = 0.5, conf_threshold: float = 0.25, iou_threshold: float = 0.5,
save_mask: bool = False, save_label: bool = False, show: bool = False, save_mask: bool = False, save_label: bool = False, show: bool = False,
@ -685,15 +717,32 @@ class YOLOSegmentationInference:
print(f"找到 {len(image_files)} 个图片文件") print(f"找到 {len(image_files)} 个图片文件")
# 推送识别数据到共享目录
tmpConfig = get_conf(input_dir, user_name, pwd)
pile_dict = get_pile_dict(input_dir, user_name, pwd)
road_dict = get_road_dict(f"{tmpConfig['ip']}/{tmpConfig['share']}/{tmpConfig['excel_dir']}", user_name, pwd)
# 处理每张图片 # 处理每张图片
for idx, image_path in enumerate(image_files): for idx, image_path in enumerate(image_files):
# for image_path in image_files: # 检查路面类型，切换模型
if not self.model :
model_conf_threshold, model_path = self.get_model_info(pile_dict, road_dict, image_path)
try:
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"设备情况：{device}")
except Exception as e:
print(f"获取设备失败: {e}")
self.set_model_info(model_path, device, model_conf_threshold)
# 加载模型
if not self.load_model():
return
result = self.process_single_image_share_dir( result = self.process_single_image_share_dir(
image_path=image_path, image_path=image_path,
user_name=user_name, user_name=user_name,
pwd=pwd, pwd=pwd,
output_dir=tmp_output_dir, output_dir=tmp_output_dir,
conf_threshold=conf_threshold, conf_threshold=self.conf_threshold,
iou_threshold=iou_threshold, iou_threshold=iou_threshold,
save_mask=save_mask, save_mask=save_mask,
save_label=save_label, save_label=save_label,
@ -716,17 +765,16 @@ class YOLOSegmentationInference:
mem_usage = memory_info.rss / 1024 / 1024 # 返回MB mem_usage = memory_info.rss / 1024 / 1024 # 返回MB
print(f"清理后，内存使用: {mem_usage:.2f} MB") print(f"清理后，内存使用: {mem_usage:.2f} MB")
# 推送识别数据到共享目录 # 处理输出目录的文件格式
tmpConfig = get_conf(input_dir, user_name, pwd)
pile_dict = get_pile_dict(input_dir, user_name, pwd)
road_dict = get_road_dict(f"{tmpConfig['ip']}/{tmpConfig['share']}/{tmpConfig['excel_dir']}", user_name, pwd)
process_dir(road_dict, pile_dict, tmp_output_dir) process_dir(road_dict, pile_dict, tmp_output_dir)
# 识别生成的文件压缩成zip # 识别生成的文件压缩成zip
zip_folder_shutil(tmp_output_dir) zip_folder_shutil(tmp_output_dir)
move_file_shutil(tmp_output_dir+".zip", tmp_output_dir + f"/{task_id}.zip") move_file_shutil(tmp_output_dir+".zip", tmp_output_dir + f"/{task_id}.zip")
remote_dir = f"{tmpConfig['dir']}_识别/{task_id}/{current_time}" remote_dir = f"{tmpConfig['dir']}_识别/{task_id}/{current_time}"
scanner.upload_directory(tmp_output_dir, config['share'], remote_dir=remote_dir) scanner.upload_directory(tmp_output_dir, config['share'], remote_dir=remote_dir)
del_file_shutil(tmp_output_dir) # del_file_shutil(tmp_output_dir)
# 更新识别任务状态为已结束 # 更新识别任务状态为已结束
DB_CONFIG = { DB_CONFIG = {
"dbname": "postgres", "dbname": "postgres",
@ -896,33 +944,8 @@ def predict_images(pt_name, zip_url, output_dir="predictions", conf_threshold=0.
def predict_images_share_dir(task_id, pt_name, zip_url, user_name, pwd, output_dir="predictions", conf_threshold=0.25, save_json=False): def predict_images_share_dir(task_id, pt_name, zip_url, user_name, pwd, output_dir="predictions", conf_threshold=0.25, save_json=False):
# 本地测试模式 - 请根据实际情况修改以下路径
# local_model_path = r"D:\project\verification\ultralytics-main\model\script\seg\pt\test.pttest.pt"
local_model_path = r"../pt_save/road_crack.pt"
local_output_dir = output_dir
# zip_url = "meta_data/ai_train_platform/train.zip"
try:
# 加载模型
print(f"正在加载模型: {local_model_path}")
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"模型已加载到 {device}")
except Exception as e:
print(f"处理目录失败: {e}")
inference = YOLOSegmentationInference(
model_path=local_model_path,
device=device
)
# 加载模型
if not inference.load_model():
return
# zip_url = r"D:\project\verification\ultralytics-main\model\script\seg\test_seg_pic"
inference = YOLOSegmentationInference()
result_save = [] result_save = []
conf_threshold = 0.25 conf_threshold = 0.25
iou_threshold = 0.5 iou_threshold = 0.5

38
util/smb_tool.py
View File

@ -1050,7 +1050,7 @@ def main():
# for i, path in enumerate(imgPaths, 1): # for i, path in enumerate(imgPaths, 1):
# print(f"{i}. {path}") # print(f"{i}. {path}")
# # 读取excel # 读取excel
# full_path = scanner.build_full_path(share_path=config['share'], file_path='西南计算机\\AA县\\每公里指标明细表（北碚）.xls') # full_path = scanner.build_full_path(share_path=config['share'], file_path='西南计算机\\AA县\\每公里指标明细表（北碚）.xls')
# df = scanner.read_excel(full_path) # df = scanner.read_excel(full_path)
# rows = scanner.process_all_rows(df) # rows = scanner.process_all_rows(df)
@ -1099,23 +1099,25 @@ def main():
# str = '\\\\192.168.110.114\\share_File\\西南计算机/北碚报送数据/3.25/图像类\\C071500109B\\Images_识别/77/20251120172316/77.zip' # str = '\\\\192.168.110.114\\share_File\\西南计算机/北碚报送数据/3.25/图像类\\C071500109B\\Images_识别/77/20251120172316/77.zip'
# print(f"standardized_path(str)={standardized_path(str)}") # print(f"standardized_path(str)={standardized_path(str)}")
road_dict_for_width = {} # road_dict_for_width = {}
df = pd.read_excel('D:/devForBdzlWork/ai-train_platform/predictions/C005500155A/nl_yy_glzb (2)_识别宽度数据读取.xls') # df = pd.read_excel('D:/devForBdzlWork/ai-train_platform/predictions/C005500155A/nl_yy_glzb (2)_识别宽度数据读取.xls')
rows = scanner.process_all_rows(df, columns=['线路编码','区划代码','方向(上行/下行)','识别宽度（米）']) # rows = scanner.process_all_rows(df, columns=['线路编码','区划代码','方向(上行/下行)','识别宽度（米）'])
for i, row in enumerate(rows, 1): # for i, row in enumerate(rows, 1):
data = row['data'] # data = row['data']
if pd.notna(data['线路编码']) : # if pd.notna(data['线路编码']) :
up_or_down = 'A' # up_or_down = 'A'
if data['方向(上行/下行)'] == '下行' : # if data['方向(上行/下行)'] == '下行' :
up_or_down = 'B' # up_or_down = 'B'
key = f"{data['线路编码']}{str(int(data['区划代码']))}{up_or_down}" # key = f"{data['线路编码']}{str(int(data['区划代码']))}{up_or_down}"
width = data.get('识别宽度（米）') # width = data.get('识别宽度（米）')
if width : # if width :
road_dict_for_width[key] = width # road_dict_for_width[key] = width
print('读取excel') # print('读取excel')
road_dict = {} road_dict = {}
df = pd.read_excel('D:/devForBdzlWork/ai-train_platform/predictions/C005500155A/每公里指标明细表-农村公路技术状况评定结果表-500155.xls') # df = pd.read_excel('D:/devForBdzlWork/ai-train_platform/predictions/C005500155A/每公里指标明细表-农村公路技术状况评定结果表-500155.xls')
# df = pd.read_csv('D:/devForBdzlWork/ai-train_platform/predictions/jlp/20260227090742/公里指标数据导出-jlp.csv')
df = pd.read_excel('D:/devForBdzlWork/ai-train_platform/predictions/jlp/20260227090742/每公里指标明细表-jlp.xlsx')
rows = scanner.process_all_rows(df, columns=['线路编码','区划代码','识别宽度（米）','方向(上行/下行)','技术等级','路面类型（沥青/水泥/砂石）','起桩号（米）','止桩号（米）']) rows = scanner.process_all_rows(df, columns=['线路编码','区划代码','识别宽度（米）','方向(上行/下行)','技术等级','路面类型（沥青/水泥/砂石）','起桩号（米）','止桩号（米）'])
for i, row in enumerate(rows, 1): for i, row in enumerate(rows, 1):
data = row['data'] data = row['data']
@ -1124,14 +1126,14 @@ def main():
if data['方向(上行/下行)'] == '下行' : if data['方向(上行/下行)'] == '下行' :
up_or_down = 'B' up_or_down = 'B'
key = f"{data['线路编码']}{str(int(data['区划代码']))}{up_or_down}" key = f"{data['线路编码']}{str(int(data['区划代码']))}{up_or_down}"
width = road_dict_for_width.get(key) width = road_dict.get(key)
if width : if width :
data['识别宽度（米）'] = width data['识别宽度（米）'] = width
if road_dict.get(key) : if road_dict.get(key) :
road_dict[key].append(row) road_dict[key].append(row)
else : else :
road_dict[key] = [row] road_dict[key] = [row]
print('读取excel') print('读取excel/csv')
if __name__ == "__main__": if __name__ == "__main__":
main() main()

112
util/yolo2pix_new.py
View File

@ -69,9 +69,14 @@ def draw_grid_on_image(image_path, grid_cells, cell_size=(GRID_WIDTH, GRID_HEIGH
overlay = image.copy() overlay = image.copy()
for cname, nums in grid_cells.items(): for cname, nums in grid_cells.items():
color = (np.random.randint(64,255),np.random.randint(64,255),np.random.randint(64,255)) color = (np.random.randint(64,255),np.random.randint(64,255),np.random.randint(64,255))
for num in nums[0]: for cell in nums :
covered_cells = cell.get("covered_cells")
for num in covered_cells:
x1,y1,x2,y2 = num_to_coord(num, cols, cell_w, cell_h) x1,y1,x2,y2 = num_to_coord(num, cols, cell_w, cell_h)
cv2.rectangle(overlay,(x1,y1),(x2,y2),color,-1) cv2.rectangle(overlay,(x1,y1),(x2,y2),color,-1)
# for num in nums[0]:
# x1,y1,x2,y2 = num_to_coord(num, cols, cell_w, cell_h)
# cv2.rectangle(overlay,(x1,y1),(x2,y2),color,-1)
cv2.addWeighted(overlay,0.4,image,0.6,0,image) cv2.addWeighted(overlay,0.4,image,0.6,0,image)
for i in range(0, w, cell_w): for i in range(0, w, cell_w):
cv2.line(image,(i,0),(i,h),(100,100,100),1) cv2.line(image,(i,0),(i,h),(100,100,100),1)
@ -95,15 +100,17 @@ def detect_road_type_from_content(label_file):
if kw in content: return "gravel" if kw in content: return "gravel"
return "gravel" return "gravel"
def get_road_info(summary_data, road_dict, pile_dict, img_file_name): def get_road_info(road_dict, pile_dict, img_file_name):
"""获取路线信息""" """获取路线信息"""
parts = pile_dict.get(img_file_name) parts = pile_dict.get(img_file_name)
if parts : if parts :
road_code = parts[0] road_code = parts[0]
road_info = road_dict.get(road_code) road_info = road_dict.get(road_code)
if road_info : if road_info :
pile_no_min = convert_special_format(summary_data[0][0]) * 1000 pile_dict_vals = list(pile_dict.values())
pile_no_max = convert_special_format(summary_data[len(summary_data)-1][0]) * 1000 pile_dict_vals_len = len(pile_dict_vals)
pile_no_min = convert_special_format(pile_dict_vals[0][1]) * 1000
pile_no_max = convert_special_format(pile_dict_vals[pile_dict_vals_len-1][1]) * 1000
if pile_no_max < pile_no_min : #上行下行 if pile_no_max < pile_no_min : #上行下行
tmp = pile_no_max tmp = pile_no_max
pile_no_max = pile_no_min pile_no_max = pile_no_min
@ -216,9 +223,19 @@ def yoloseg_to_grid_share_dir(road_dict,pile_dict,image_path,label_file,grid_wid
pile_no_tmp = f"桩号:K000{pile_no}" pile_no_tmp = f"桩号:K000{pile_no}"
result_lines.append(f"{cname} {pile_no_tmp} {ROAD_TYPE_EN_TO_CN.get(road_type, 'xx')} {ids_str}") result_lines.append(f"{cname} {pile_no_tmp} {ROAD_TYPE_EN_TO_CN.get(road_type, 'xx')} {ids_str}")
if cname not in all_class_cells: all_class_cells[cname]=set() if cname not in all_class_cells: all_class_cells[cname]=set()
cell_info.append(covered_cells) # cell_info.append(covered_cells)
cell_info.append([max_x - min_x, max_y - min_y]) # cell_info.append([max_x - min_x, max_y - min_y])
# all_class_cells[cname] = cell_info
if all_class_cells[cname] :
cell_info = all_class_cells[cname]
cell_info.append({
"covered_cells":covered_cells,
"width": max_x - min_x,
"height": max_y - min_y
})
all_class_cells[cname] = cell_info all_class_cells[cname] = cell_info
return '\n'.join(result_lines), all_class_cells, road_type, rows * cols return '\n'.join(result_lines), all_class_cells, road_type, rows * cols
def yoloseg_to_grid(image_path,label_file,cover_ratio=COVER_RATIO): def yoloseg_to_grid(image_path,label_file,cover_ratio=COVER_RATIO):
@ -323,11 +340,11 @@ def in_interval(increment, cur_pile_no, tmp_start, tmp_end) :
return 0 return 0
# 指定间隔区间，输出基本信息+灾害数据 # 指定间隔区间，输出基本信息+灾害数据
def process_info(road_dict,pile_dict,summary_data,image_path,current_time,interval=10,dir="output",cell_area=CELL_AREA,grid_width=GRID_WIDTH,grid_height=GRID_HEIGHT) : def process_info(road_dict,pile_dict,summary_data,current_time,interval=10,dir="output",cell_area=CELL_AREA,grid_width=GRID_WIDTH,grid_height=GRID_HEIGHT) :
process_info_data = [] process_info_data = []
if summary_data: if summary_data:
img_file_name = os.path.basename(image_path) img_file_name = list(pile_dict.keys())[0]
road_data = get_road_info(summary_data, road_dict, pile_dict, img_file_name) road_data = get_road_info(road_dict, pile_dict, img_file_name)
identify_width = road_data.get('识别宽度（米）', '3.6') identify_width = road_data.get('识别宽度（米）', '3.6')
up_or_down = road_data.get('方向(上行/下行)', '上行') up_or_down = road_data.get('方向(上行/下行)', '上行')
road_code = pile_dict.get(img_file_name)[0] road_code = pile_dict.get(img_file_name)[0]
@ -431,7 +448,8 @@ def process_info(road_dict,pile_dict,summary_data,image_path,current_time,interv
break break
# 同列汇总 10m一个区间--对应5张图 # 同列汇总 10m一个区间--对应5张图
column_sums = [f"{(sum(column)/(interval / 2)):0.2f}" for column in zip(*subRows)] tmp_interval = round((tmp_end - tmp_start) * 1000)
column_sums = [f"{(sum(column)/(tmp_interval / 2)):0.2f}" for column in zip(*subRows)]
row += column_sums row += column_sums
# f.write(','.join(row)+'\n') # f.write(','.join(row)+'\n')
row += ['','','','','','','','','','','','','','',''] row += ['','','','','','','','','','','','','','','']
@ -541,6 +559,7 @@ def process_dir(road_dict,pile_dict,dir="output",cell_area=CELL_AREA,cell_width=
# 解压 # 解压
# 读取桩号映射 # 读取桩号映射
# 遍历图片 # 遍历图片
image_path = None
summary_data = [] summary_data = []
for root,_,files in os.walk(dir): for root,_,files in os.walk(dir):
for file in files: for file in files:
@ -556,14 +575,17 @@ def process_dir(road_dict,pile_dict,dir="output",cell_area=CELL_AREA,cell_width=
with open(grid_txt_path,'w',encoding='utf-8') as f: with open(grid_txt_path,'w',encoding='utf-8') as f:
f.write(out_txt) f.write(out_txt)
# 生成网格可视化 # 生成网格可视化
# draw_grid_on_image(image_path,class_cells,cell_size=(GRID_WIDTH, GRID_HEIGHT),save_path=os.path.splitext(image_path)[0]+"_grid.jpg") draw_grid_on_image(image_path,class_cells,cell_size=(GRID_WIDTH, GRID_HEIGHT),save_path=os.path.splitext(image_path)[0]+"_grid.jpg")
# 统计各类面积 # 统计各类面积
counts = {k:[len(v[0])*cell_area, v[1][0], v[1][1]] for k,v in class_cells.items()} # counts = {k:[len(v[0])*cell_area, v[1][0], v[1][1]] for k,v in class_cells.items()}
counts = {k:[[len(v_child.get("covered_cells"))*cell_area, v_child.get("width"), v_child.get("height")] for v_child in v] for k,v in class_cells.items()}
# total_area = sum(counts.values()) # total_area = sum(counts.values())
# 灾害总面积比例 # 灾害总面积比例
merged_set = set([]) merged_set = set([])
for k,v in class_cells.items() : for k,v in class_cells.items() :
merged_set = merged_set.union(v[0]) # merged_set = merged_set.union(v[0])
for v_child in v:
merged_set = merged_set.union(v_child.get("covered_cells"))
total_area = len(merged_set) total_area = len(merged_set)
# 桩号 路线编号 # 桩号 路线编号
@ -580,24 +602,24 @@ def process_dir(road_dict,pile_dict,dir="output",cell_area=CELL_AREA,cell_width=
current_time = datetime.now().strftime("%Y%m%d%H%M%S") current_time = datetime.now().strftime("%Y%m%d%H%M%S")
# 写桩号问题列表.txt # 写桩号问题列表.txt
process_damage_detail_txt(road_dict, pile_dict, dir, summary_data, image_path, current_time) process_damage_detail_txt(road_dict, pile_dict, dir, summary_data, current_time)
# 灾害数据.txt # 灾害数据.txt
process_damage_txt(road_dict, pile_dict, dir, summary_data, image_path, current_time) process_damage_txt(road_dict, pile_dict, dir, summary_data, current_time)
# 病害明细列表.xlsx # 病害明细列表.xlsx
img_file_path = process_damage_detail_excel(road_dict, pile_dict, dir, cell_area, cell_width, cell_height, summary_data, image_path) process_damage_detail_excel(road_dict, pile_dict, dir, cell_area, cell_width, cell_height, summary_data)
# 综合明细表.xlsx # 综合明细表.xlsx
process_damage_composite_excel(road_dict, pile_dict, summary_data, image_path, current_time, img_file_path) process_damage_composite_excel(road_dict, pile_dict, dir, summary_data, current_time)
def process_damage_composite_excel(road_dict, pile_dict, summary_data, image_path, current_time, img_file_path): def process_damage_composite_excel(road_dict, pile_dict, dir, summary_data, current_time):
print("输出:综合明细表.xlsx") print("输出:综合明细表.xlsx")
heads = ['路线编码','起点','终点','车道编码','上下行','公路等级','路面类型','PQI','PQI等级','DR(%)','PCI','PCI等级','IRI','RQI','RQI等级','RD','RDI','RDI等级','SMTD','PBI','PBI等级','WR','PWI','PWI等级','备注'] heads = ['路线编码','起点','终点','车道编码','上下行','公路等级','路面类型','PQI','PQI等级','DR(%)','PCI','PCI等级','IRI','RQI','RQI等级','RD','RDI','RDI等级','SMTD','PBI','PBI等级','WR','PWI','PWI等级','备注']
data1 = process_info(road_dict,pile_dict,summary_data,image_path,current_time,10) data1 = process_info(road_dict,pile_dict,summary_data,current_time,10)
data2 = process_info(road_dict,pile_dict,summary_data,image_path,current_time,100) data2 = process_info(road_dict,pile_dict,summary_data,current_time,100)
data3 = process_info(road_dict,pile_dict,summary_data,image_path,current_time,1000) data3 = process_info(road_dict,pile_dict,summary_data,current_time,1000)
excel_data = [ excel_data = [
{ {
@ -619,13 +641,13 @@ def process_damage_composite_excel(road_dict, pile_dict, summary_data, image_pat
'data': data3 'data': data3
} }
] ]
create_multiple_sheets_with_multiple_headers(img_file_path + '/excel/综合明细表.xlsx', excel_data) create_multiple_sheets_with_multiple_headers(f"{dir}/excel/综合明细表.xlsx", excel_data)
def process_damage_detail_txt(road_dict, pile_dict, dir, summary_data, image_path, current_time): def process_damage_detail_txt(road_dict, pile_dict, dir, summary_data, current_time):
if summary_data: if summary_data:
img_file_name = os.path.basename(image_path) img_file_name = list(pile_dict.keys())[0]
road_data = get_road_info(summary_data, road_dict, pile_dict, img_file_name) road_data = get_road_info(road_dict, pile_dict, img_file_name)
road_code = pile_dict.get(img_file_name)[0] road_code = pile_dict.get(img_file_name)[0]
road_type = summary_data[0][3] road_type = summary_data[0][3]
@ -645,20 +667,22 @@ def process_damage_detail_txt(road_dict, pile_dict, dir, summary_data, image_pat
else: else:
keys = list(CLASS_MAP_GRAVEL.keys()) keys = list(CLASS_MAP_GRAVEL.keys())
for k in keys: for k in keys:
row.append(f"{counts.get(k,[0,0,0])[0]:.2f}") # row.append(f"{counts.get(k,[0,0,0])[0]:.2f}")
sum_count = sum(count[0] for count in counts.get(k, [[0, 0, 0]]))
row.append(f"{sum_count:.2f}")
f.write(','.join(row)+'\n') f.write(','.join(row)+'\n')
print(f"输出完成: {out_file}") print(f"输出完成: {out_file}")
def process_damage_detail_excel(road_dict, pile_dict, dir, cell_area, cell_width, cell_height, summary_data, image_path): def process_damage_detail_excel(road_dict, pile_dict, dir, cell_area, cell_width, cell_height, summary_data):
print("输出:病害明细列表.xlsx") print("输出:病害明细列表.xlsx")
os.makedirs(f"{dir}/excel", exist_ok=True) os.makedirs(f"{dir}/excel", exist_ok=True)
headers = ['序号','路线编码','方向','桩号','路面类型','病害名称','程度','长度(m)',' 宽度(m)',' 面积(㎡)',' 横向位置','备注'] headers = ['序号','路线编码','方向','桩号','路面类型','病害名称','程度','长度(m)',' 宽度(m)',' 面积(㎡)',' 横向位置','备注']
data_list = [] data_list = []
if summary_data: if summary_data:
img_file_path = os.path.dirname(image_path) img_file_name = list(pile_dict.keys())[0]
img_file_name = os.path.basename(image_path) road_data = get_road_info(road_dict, pile_dict, img_file_name)
road_data = get_road_info(summary_data, road_dict, pile_dict, img_file_name)
road_code, pile_no, road_type = detect_road_type_from_road_dict(road_dict, pile_dict, img_file_name) road_code, pile_no, road_type = detect_road_type_from_road_dict(road_dict, pile_dict, img_file_name)
identify_width = road_data.get('识别宽度（米）', '3.6') identify_width = road_data.get('识别宽度（米）', '3.6')
up_or_down = road_data.get('方向(上行/下行)', '上行') up_or_down = road_data.get('方向(上行/下行)', '上行')
@ -666,18 +690,18 @@ def process_damage_detail_excel(road_dict, pile_dict, dir, cell_area, cell_width
for data in summary_data: for data in summary_data:
damage_data = data[2] damage_data = data[2]
for attr_name, attr_value in damage_data.items(): for attr_name, attr_value in damage_data.items():
excel_data = [excel_index, road_code, up_or_down, f"K000{data[0]}", ROAD_TYPE_EN_TO_CN.get(road_type), attr_name, '', attr_value[1]*cell_width, attr_value[2]*cell_height, attr_value[0], '', ''] for attr_value_child in attr_value :
excel_data = [excel_index, road_code, up_or_down, f"K000{data[0]}", ROAD_TYPE_EN_TO_CN.get(road_type), attr_name, '', attr_value_child[1]*cell_width, attr_value_child[2]*cell_height, attr_value_child[0], '', '']
data_list.append(excel_data) data_list.append(excel_data)
all_data = [headers] + data_list all_data = [headers] + data_list
smb_tool.write_to_excel_pandas(all_data, img_file_path + '/excel/病害明细列表.xlsx') smb_tool.write_to_excel_pandas(all_data, f"{dir}/excel/病害明细列表.xlsx")
return img_file_path
def process_damage_txt(road_dict, pile_dict, dir, summary_data, image_path, current_time): def process_damage_txt(road_dict, pile_dict, dir, summary_data, current_time):
if summary_data: if summary_data:
img_file_name = os.path.basename(image_path) img_file_name = list(pile_dict.keys())[0]
road_data = get_road_info(summary_data, road_dict, pile_dict, img_file_name) road_data = get_road_info(road_dict, pile_dict, img_file_name)
identify_width = road_data.get('识别宽度（米）', '3.6') identify_width = road_data.get('识别宽度（米）', '3.6')
up_or_down = road_data.get('方向(上行/下行)', '上行') up_or_down = road_data.get('方向(上行/下行)', '上行')
road_code = pile_dict.get(img_file_name)[0] road_code = pile_dict.get(img_file_name)[0]
@ -780,14 +804,17 @@ def process_damage_txt(road_dict, pile_dict, dir, summary_data, image_path, curr
else: else:
keys = list(CLASS_MAP_GRAVEL.keys()) keys = list(CLASS_MAP_GRAVEL.keys())
for k in keys: for k in keys:
tmp_row.append(counts.get(k, [0,0,0])[0]) # tmp_row.append(counts.get(k, [0,0,0])[0])
sum_count = sum(count[0] for count in counts.get(k, [[0, 0, 0]]))
tmp_row.append(sum_count)
subRows.append(tmp_row) subRows.append(tmp_row)
index = index + 1 index = index + 1
else : else :
break break
# 同列汇总 10m一个区间--对应5张图 # 同列汇总 10m一个区间--对应5张图
column_sums = [f"{(sum(column)/5):0.2f}" for column in zip(*subRows)] tmp_interval = round((tmp_end - tmp_start) * 1000)
column_sums = [f"{(sum(column)/tmp_interval):0.2f}" for column in zip(*subRows)]
row += column_sums row += column_sums
f.write(','.join(row)+'\n') f.write(','.join(row)+'\n')
else : else :
@ -879,10 +906,11 @@ def get_road_dict(local_dir):
dict: 路线编码到路线信息的映射字典 dict: 路线编码到路线信息的映射字典
""" """
# 查找匹配的Excel文件 # 查找匹配的Excel文件
pattern = os.path.join(local_dir, '每公里指标明细表*.xls') pattern_xls = os.path.join(local_dir, '每公里指标明细表*.xls')
found_paths = glob.glob(pattern) pattern_xlsx = os.path.join(local_dir, '每公里指标明细表*.xlsx')
found_paths = glob.glob(pattern_xls) + glob.glob(pattern_xlsx)
print(f"\n找到 {len(found_paths)} 个 '每公里指标明细表*.xls' 文件:") print(f"\n找到 {len(found_paths)} 个 '每公里指标明细表*.xls/xlsx' 文件:")
for i, path in enumerate(found_paths, 1): for i, path in enumerate(found_paths, 1):
print(f"{i}. {path}") print(f"{i}. {path}")
@ -961,7 +989,7 @@ if __name__=="__main__":
# calc_cell_area, calc_grid_width, calc_grid_height = calc_grid_param(2048, 4096, 3.6, 2) # calc_cell_area, calc_grid_width, calc_grid_height = calc_grid_param(2048, 4096, 3.6, 2)
# print(f"calc_cell_area={calc_cell_area}, calc_grid_width={calc_grid_width}, calc_grid_height={calc_grid_height}") # print(f"calc_cell_area={calc_cell_area}, calc_grid_width={calc_grid_width}, calc_grid_height={calc_grid_height}")
output_dir = "D:/devForBdzlWork/ai-train_platform/predictions/C234500155A" output_dir = "D:/devForBdzlWork/ai-train_platform/predictions/jlp/20260227090742"
pile_dict = get_pile_dict(output_dir) pile_dict = get_pile_dict(output_dir)
road_dict = get_road_dict(output_dir) road_dict = get_road_dict(output_dir)
process_dir(road_dict, pile_dict, output_dir) process_dir(road_dict, pile_dict, output_dir)