File size: 3,641 Bytes
ec8729d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
# %%
import json
import os
import sys
import cv2
import logging
from logging.handlers import RotatingFileHandler
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
# 配置日志记录到文件
# 日志配置
def setup_logger(log_file):
logger = logging.getLogger("ImageProcessingLogger")
logger.setLevel(logging.DEBUG)
# 设置文件滚动，每个日志文件最大10MB，保留5个文件
handler = RotatingFileHandler(log_file, maxBytes=10 * 1024 * 1024, backupCount=5)
handler.setLevel(logging.DEBUG)
# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
# 添加处理器
logger.addHandler(handler)
return logger
# 日志实例
logger = setup_logger("app.log")
def read_json(file_path):
with open(file_path, 'r', encoding='utf-8') as file:
data = json.load(file)
return data
def write_json(file_path, data):
with open(file_path, 'w', encoding='utf-8') as file:
json.dump(data, file, ensure_ascii=False, indent=4)
def get_corrdinate(data, image_path,logger, out_root=None):
try:
image_name = image_path.split('/')[-1]
out_path = os.path.join(out_root, image_name)
y_str, x_str = data['click_loc'].split(',')[2:]
y = float(y_str.split(':')[1])
x = float(x_str.strip())
# print(f"x: {x}, y: {y}")
image = cv2.imread(image_path)
if image is None:
logger.error(f"Failed to read image: {image_path}")
return
height, width, _ = image.shape
# print(f"height: {height}, width: {width}")
x = int(x * width)
y = int(y * height)
# print(f"x: {x}, y: {y}")
color = (0, 255, 0)
thickness = 5
center = (x, y)
length = 35
cv2.line(image, (center[0] - length, center[1] - length), (center[0] - length, center[1] + length), color, thickness)
cv2.line(image, (center[0] - length, center[1] + length), (center[0] + length, center[1] + length), color, thickness)
cv2.line(image, (center[0] + length, center[1] + length), (center[0] + length, center[1] - length), color, thickness)
cv2.line(image, (center[0] + length, center[1] - length), (center[0] - length, center[1] - length), color, thickness)
cv2.imwrite(out_path, image)
# 每处理100张记录日志
if count % 100 == 0:
logger.info(f"Processed {count} images successfully.")
except Exception as e:
logger.exception(f"Error processing data: {data}. Exception: {e}")
def process_data(data, root_path,logger, out_root):
image_path = os.path.join(root_path, data['image'])
get_corrdinate(data, image_path,logger, out_root)
if __name__ == "__main__":
# 参数配置
json_file = r'/code/Auto-GUI/dataset/mind/general_blip_train_llava_coco.json'
root_path = r'/code/Auto-GUI/dataset'
out_root = r'/code/Auto-GUI/dataset/coco_corrdinate'
count = 0
# 读取 JSON 数据
data = read_json(json_file)
data1 = [line for line in data if line['action_type'] == '#DUAL_POINT#'][1:]
logging.debug(f"total data: " +str(len(data1)))
# 并行处理数据
os.makedirs(out_root, exist_ok=True)
with ThreadPoolExecutor(max_workers=8) as executor:
list(tqdm(executor.map(lambda d: process_data(d, root_path, logger, out_root), data1), total=len(data1)))
# list(tqdm(executor.map(lambda d: process_data(d, root_path, out_root), data1), total=len(data1)))
logger.info("All processing complete.")
|