get_coordinate.py

# %%
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.")