Portrait-Infer / ensemble_results.py

Upload ensemble_results.py

11d7bc2 verified 5 months ago

10 kB

import os
import re
import json
import glob
import copy
from collections import Counter, defaultdict
from statistics import median

CRITERIA = [
    "Color Harmony",
    "Visual Style Consistency",
    "Sharpness",
    "Light and Shadow Modeling",
    "Creativity and Originality",
    "Exposure Control",
    "Application of Classical Composition Principles",
    "Depth of Field and Layering",
    "Visual Center Stability",
    "Visual Flow Guidance",
    "Structural Support Stability",
    "Appropriateness of Negative Space",
    "Subject Integrity",
]

# LEVEL_ORDER = {"Poor": 0, "Medium": 1, "Good": 2}
# LEVEL_INV = {0: "Poor", 1: "Medium", 2: "Good"}


LEVEL_ORDER = {"Poor": 0, "Medium": 1, "Good": 2}
LEVEL_INV = {0: "A", 1: "B", 2: "C"}


def normalize_level(x):
    if not isinstance(x, str):
        return None
    x = x.strip().lower()
    mp = {
        "poor": "Poor",
        "medium": "Medium",
        "good": "Good",
    }
    return mp.get(x)


def basename_from_item(item):
    img_path = item.get("images", [{}])[0].get("path", "")
    return os.path.basename(img_path)


def parse_response_raw(resp):
    """
    支持：
    - "{\"total_score\": 84}"
    - "41"
    - "{\"criteria\": {...}}"
    - "[\"Medium\", ...]"
    - "{\"answer\": \"C\"}"
    """
    if isinstance(resp, (dict, list, int, float)):
        return resp

    if not isinstance(resp, str):
        return None

    s = resp.strip()

    # 纯数字分数
    if re.fullmatch(r"-?\d+(\.\d+)?", s):
        return float(s)

    # 标准 JSON
    try:
        return json.loads(s)
    except Exception:
        pass

    # 兜底：提取 {...}
    m = re.search(r"\{.*\}", s, flags=re.S)
    if m:
        try:
            return json.loads(m.group(0))
        except Exception:
            pass

    # 兜底：提取 [...]
    m = re.search(r"\[.*\]", s, flags=re.S)
    if m:
        try:
            return json.loads(m.group(0))
        except Exception:
            pass

    return None


def iter_json_or_jsonl(path):
    with open(path, "r", encoding="utf-8") as f:
        text = f.read().strip()

    if not text:
        return []

    try:
        obj = json.loads(text)
        if isinstance(obj, list):
            return obj
        if isinstance(obj, dict):
            return [obj]
    except Exception:
        pass

    rows = []
    for line in text.splitlines():
        line = line.strip()
        if line:
            rows.append(json.loads(line))
    return rows


def read_all_files(folder, recursive=True):
    pattern = "**/*.json*" if recursive else "*.json*"
    files = sorted(glob.glob(os.path.join(folder, pattern), recursive=recursive))

    rows = []
    for fp in files:
        try:
            rows.extend(iter_json_or_jsonl(fp))
        except Exception as e:
            print(f"[WARN] failed to read {fp}: {e}")
    return rows


def parse_score_folder(folder):
    pred = defaultdict(list)

    for item in read_all_files(folder):
        name = basename_from_item(item)
        data = parse_response_raw(item.get("response", ""))

        score = None

        if isinstance(data, dict):
            score = data.get("total_score")
        elif isinstance(data, (int, float)):
            score = data
        elif isinstance(data, str):
            if re.fullmatch(r"-?\d+(\.\d+)?", data.strip()):
                score = float(data.strip())

        if name and score is not None:
            try:
                score = float(score)
                score = max(0, min(100, score))
                pred[name].append(score)
            except Exception:
                pass

    return pred


def parse_level_folder(folder):
    pred = defaultdict(lambda: defaultdict(list))

    for item in read_all_files(folder):
        name = basename_from_item(item)
        data = parse_response_raw(item.get("response", ""))

        if not name:
            continue

        # 格式1：{"criteria": {"Color Harmony": "Good", ...}}
        if isinstance(data, dict) and isinstance(data.get("criteria"), dict):
            criteria = data["criteria"]
            for c in CRITERIA:
                lv = normalize_level(criteria.get(c))
                if lv:
                    pred[name][c].append(lv)

        # 格式2：["Medium", "Medium", ..., 共13个]
        elif isinstance(data, list):
            for c, lv_raw in zip(CRITERIA, data):
                lv = normalize_level(lv_raw)
                if lv:
                    pred[name][c].append(lv)

    return pred


def parse_reason_folder(folder):
    pred = defaultdict(list)

    for item in read_all_files(folder):
        name = basename_from_item(item)
        data = parse_response_raw(item.get("response", ""))

        ans = None

        if isinstance(data, dict):
            ans = data.get("answer")
        elif isinstance(data, str):
            ans = data

        if name and isinstance(ans, str):
            ans = ans.strip().upper()
            if ans in {"A", "B", "C", "D"}:
                pred[name].append(ans)

    return pred


def majority_vote(values, default=None):
    values = [v for v in values if v is not None]
    if not values:
        return default

    cnt = Counter(values)

    # 平票时按第一次出现顺序
    return max(cnt.keys(), key=lambda x: (cnt[x], -values.index(x)))


def ensemble_scores(score_dicts, method="mean"):
    merged = defaultdict(list)

    # print("merged is", merged)

    for d in score_dicts:
        for name, scores in d.items():
            merged[name].extend(scores)

    out = {}
    for name, scores in merged.items():

        
        if method == "median":
            val = median(scores)
        else:
            val = sum(scores) / len(scores)

        # print("scores are", name, scores, val)

        out[name] = int(round(max(0, min(100, val))))
        # out[name] = int(round(val))

    return out



#####################

LEVEL_SCORE = {
    "Poor": 2.5,
    "Medium": 6.0,
    "Good": 8.5,
}


def score_to_level(score):
    if 0 <= score < 5:
        return "A"
    elif 5 <= score < 7:
        return "B"
    elif 7 <= score <= 10:
        return "C"
    else:
        # 兜底，防止异常值
        score = max(0, min(10, score))
        if score < 5:
            return "A"
        elif score < 7:
            return "B"
        return "C"


def ensemble_levels(level_dicts, method="score_mean"):
    merged = defaultdict(lambda: defaultdict(list))

    for d in level_dicts:
        for name, cd in d.items():
            for c, levels in cd.items():
                merged[name][c].extend(levels)

    out = defaultdict(dict)

    for name, cd in merged.items():
        for c in CRITERIA:
            vals = cd.get(c, [])
            if not vals:
                continue

            if method == "score_mean":
                nums = [LEVEL_SCORE[v] for v in vals if v in LEVEL_SCORE]
                if nums:
                    avg_score = sum(nums) / len(nums)
                    out[name][c] = score_to_level(avg_score)

            elif method == "vote":
                out[name][c] = majority_vote(vals, default="Medium")

            elif method == "ordinal_mean":
                nums = [LEVEL_ORDER[v] for v in vals if v in LEVEL_ORDER]
                if nums:
                    out[name][c] = LEVEL_INV[int(round(sum(nums) / len(nums)))]

    return out



######################


def ensemble_answers(reason_dicts):
    merged = defaultdict(list)

    for d in reason_dicts:
        for name, answers in d.items():
            merged[name].extend(answers)

    return {
        name: majority_vote(answers, default="A")
        for name, answers in merged.items()
    }


def build_submission(
    template_path,
    score_model_folders,
    level_model_folders,
    reason_model_folders,
    output_path,
    score_method="mean",
    level_method="vote",
):
    score_dicts = [parse_score_folder(p) for p in score_model_folders]
    level_dicts = [parse_level_folder(p) for p in level_model_folders]
    reason_dicts = [parse_reason_folder(p) for p in reason_model_folders]

    score_ens = ensemble_scores(score_dicts, method=score_method)
    level_ens = ensemble_levels(level_dicts, method=level_method)
    answer_ens = ensemble_answers(reason_dicts)

    with open(template_path, "r", encoding="utf-8") as f:
        result = json.load(f)

    missing_score = 0
    missing_level = 0
    missing_answer = 0

    for item in result:
        name = item["image_path"]

        if name in score_ens:
            item["total_score"] = score_ens[name]
        else:
            missing_score += 1

        for c in CRITERIA:
            if name in level_ens and c in level_ens[name]:
                item["criteria"][c]["level"] = level_ens[name][c]
            else:
                missing_level += 1

        if name in answer_ens:
            item["answer"] = answer_ens[name]
        else:
            missing_answer += 1

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)

    print(f"Saved to: {output_path}")
    print(f"Images: {len(result)}")
    print(f"Missing score images: {missing_score}")
    print(f"Missing level fields: {missing_level}")
    print(f"Missing answer images: {missing_answer}")


if __name__ == "__main__":

    # from pathlib import Path
    # ROOT = Path("/mnt/shared-storage-user/zhuxiaorong/liyunhao_data/my_code/cvpr26_challenge")


    TEMPLATE_PATH = "track_1_test_demo.json"
    OUTPUT_PATH = "./track_1_test.json"


    SCORE_MODEL_FOLDERS = [
        "./result-score"
    ]
    LEVEL_MODEL_FOLDERS = [
        "./result-level"
    ]
    REASON_MODEL_FOLDERS = [
        "./result-reason"
    ]


    build_submission(
        template_path=TEMPLATE_PATH,
        score_model_folders=SCORE_MODEL_FOLDERS,
        level_model_folders=LEVEL_MODEL_FOLDERS,
        reason_model_folders=REASON_MODEL_FOLDERS,
        output_path=OUTPUT_PATH,
        score_method="mean",
        level_method="score_mean",
    )

	import os
	import re
	import json
	import glob
	import copy
	from collections import Counter, defaultdict
	from statistics import median

	CRITERIA = [
	"Color Harmony",
	"Visual Style Consistency",
	"Sharpness",
	"Light and Shadow Modeling",
	"Creativity and Originality",
	"Exposure Control",
	"Application of Classical Composition Principles",
	"Depth of Field and Layering",
	"Visual Center Stability",
	"Visual Flow Guidance",
	"Structural Support Stability",
	"Appropriateness of Negative Space",
	"Subject Integrity",
	]

	# LEVEL_ORDER = {"Poor": 0, "Medium": 1, "Good": 2}
	# LEVEL_INV = {0: "Poor", 1: "Medium", 2: "Good"}


	LEVEL_ORDER = {"Poor": 0, "Medium": 1, "Good": 2}
	LEVEL_INV = {0: "A", 1: "B", 2: "C"}


	def normalize_level(x):
	if not isinstance(x, str):
	return None
	x = x.strip().lower()
	mp = {
	"poor": "Poor",
	"medium": "Medium",
	"good": "Good",
	}
	return mp.get(x)


	def basename_from_item(item):
	img_path = item.get("images", [{}])[0].get("path", "")
	return os.path.basename(img_path)


	def parse_response_raw(resp):
	"""
	支持：
	- "{\"total_score\": 84}"
	- "41"
	- "{\"criteria\": {...}}"
	- "[\"Medium\", ...]"
	- "{\"answer\": \"C\"}"
	"""
	if isinstance(resp, (dict, list, int, float)):
	return resp

	if not isinstance(resp, str):
	return None

	s = resp.strip()

	# 纯数字分数
	if re.fullmatch(r"-?\d+(\.\d+)?", s):
	return float(s)

	# 标准 JSON
	try:
	return json.loads(s)
	except Exception:
	pass

	# 兜底：提取 {...}
	m = re.search(r"\{.*\}", s, flags=re.S)
	if m:
	try:
	return json.loads(m.group(0))
	except Exception:
	pass

	# 兜底：提取 [...]
	m = re.search(r"\[.*\]", s, flags=re.S)
	if m:
	try:
	return json.loads(m.group(0))
	except Exception:
	pass

	return None


	def iter_json_or_jsonl(path):
	with open(path, "r", encoding="utf-8") as f:
	text = f.read().strip()

	if not text:
	return []

	try:
	obj = json.loads(text)
	if isinstance(obj, list):
	return obj
	if isinstance(obj, dict):
	return [obj]
	except Exception:
	pass

	rows = []
	for line in text.splitlines():
	line = line.strip()
	if line:
	rows.append(json.loads(line))
	return rows


	def read_all_files(folder, recursive=True):
	pattern = "*/.json" if recursive else ".json*"
	files = sorted(glob.glob(os.path.join(folder, pattern), recursive=recursive))

	rows = []
	for fp in files:
	try:
	rows.extend(iter_json_or_jsonl(fp))
	except Exception as e:
	print(f"[WARN] failed to read {fp}: {e}")
	return rows


	def parse_score_folder(folder):
	pred = defaultdict(list)

	for item in read_all_files(folder):
	name = basename_from_item(item)
	data = parse_response_raw(item.get("response", ""))

	score = None

	if isinstance(data, dict):
	score = data.get("total_score")
	elif isinstance(data, (int, float)):
	score = data
	elif isinstance(data, str):
	if re.fullmatch(r"-?\d+(\.\d+)?", data.strip()):
	score = float(data.strip())

	if name and score is not None:
	try:
	score = float(score)
	score = max(0, min(100, score))
	pred[name].append(score)
	except Exception:
	pass

	return pred


	def parse_level_folder(folder):
	pred = defaultdict(lambda: defaultdict(list))

	for item in read_all_files(folder):
	name = basename_from_item(item)
	data = parse_response_raw(item.get("response", ""))

	if not name:
	continue

	# 格式1：{"criteria": {"Color Harmony": "Good", ...}}
	if isinstance(data, dict) and isinstance(data.get("criteria"), dict):
	criteria = data["criteria"]
	for c in CRITERIA:
	lv = normalize_level(criteria.get(c))
	if lv:
	pred[name][c].append(lv)

	# 格式2：["Medium", "Medium", ..., 共13个]
	elif isinstance(data, list):
	for c, lv_raw in zip(CRITERIA, data):
	lv = normalize_level(lv_raw)
	if lv:
	pred[name][c].append(lv)

	return pred


	def parse_reason_folder(folder):
	pred = defaultdict(list)

	for item in read_all_files(folder):
	name = basename_from_item(item)
	data = parse_response_raw(item.get("response", ""))

	ans = None

	if isinstance(data, dict):
	ans = data.get("answer")
	elif isinstance(data, str):
	ans = data

	if name and isinstance(ans, str):
	ans = ans.strip().upper()
	if ans in {"A", "B", "C", "D"}:
	pred[name].append(ans)

	return pred


	def majority_vote(values, default=None):
	values = [v for v in values if v is not None]
	if not values:
	return default

	cnt = Counter(values)

	# 平票时按第一次出现顺序
	return max(cnt.keys(), key=lambda x: (cnt[x], -values.index(x)))


	def ensemble_scores(score_dicts, method="mean"):
	merged = defaultdict(list)

	# print("merged is", merged)

	for d in score_dicts:
	for name, scores in d.items():
	merged[name].extend(scores)

	out = {}
	for name, scores in merged.items():


	if method == "median":
	val = median(scores)
	else:
	val = sum(scores) / len(scores)

	# print("scores are", name, scores, val)

	out[name] = int(round(max(0, min(100, val))))
	# out[name] = int(round(val))

	return out



	#####################

	LEVEL_SCORE = {
	"Poor": 2.5,
	"Medium": 6.0,
	"Good": 8.5,
	}


	def score_to_level(score):
	if 0 <= score < 5:
	return "A"
	elif 5 <= score < 7:
	return "B"
	elif 7 <= score <= 10:
	return "C"
	else:
	# 兜底，防止异常值
	score = max(0, min(10, score))
	if score < 5:
	return "A"
	elif score < 7:
	return "B"
	return "C"


	def ensemble_levels(level_dicts, method="score_mean"):
	merged = defaultdict(lambda: defaultdict(list))

	for d in level_dicts:
	for name, cd in d.items():
	for c, levels in cd.items():
	merged[name][c].extend(levels)

	out = defaultdict(dict)

	for name, cd in merged.items():
	for c in CRITERIA:
	vals = cd.get(c, [])
	if not vals:
	continue

	if method == "score_mean":
	nums = [LEVEL_SCORE[v] for v in vals if v in LEVEL_SCORE]
	if nums:
	avg_score = sum(nums) / len(nums)
	out[name][c] = score_to_level(avg_score)

	elif method == "vote":
	out[name][c] = majority_vote(vals, default="Medium")

	elif method == "ordinal_mean":
	nums = [LEVEL_ORDER[v] for v in vals if v in LEVEL_ORDER]
	if nums:
	out[name][c] = LEVEL_INV[int(round(sum(nums) / len(nums)))]

	return out



	######################


	def ensemble_answers(reason_dicts):
	merged = defaultdict(list)

	for d in reason_dicts:
	for name, answers in d.items():
	merged[name].extend(answers)

	return {
	name: majority_vote(answers, default="A")
	for name, answers in merged.items()
	}


	def build_submission(
	template_path,
	score_model_folders,
	level_model_folders,
	reason_model_folders,
	output_path,
	score_method="mean",
	level_method="vote",
	):
	score_dicts = [parse_score_folder(p) for p in score_model_folders]
	level_dicts = [parse_level_folder(p) for p in level_model_folders]
	reason_dicts = [parse_reason_folder(p) for p in reason_model_folders]

	score_ens = ensemble_scores(score_dicts, method=score_method)
	level_ens = ensemble_levels(level_dicts, method=level_method)
	answer_ens = ensemble_answers(reason_dicts)

	with open(template_path, "r", encoding="utf-8") as f:
	result = json.load(f)

	missing_score = 0
	missing_level = 0
	missing_answer = 0

	for item in result:
	name = item["image_path"]

	if name in score_ens:
	item["total_score"] = score_ens[name]
	else:
	missing_score += 1

	for c in CRITERIA:
	if name in level_ens and c in level_ens[name]:
	item["criteria"][c]["level"] = level_ens[name][c]
	else:
	missing_level += 1

	if name in answer_ens:
	item["answer"] = answer_ens[name]
	else:
	missing_answer += 1

	with open(output_path, "w", encoding="utf-8") as f:
	json.dump(result, f, ensure_ascii=False, indent=2)

	print(f"Saved to: {output_path}")
	print(f"Images: {len(result)}")
	print(f"Missing score images: {missing_score}")
	print(f"Missing level fields: {missing_level}")
	print(f"Missing answer images: {missing_answer}")


	if __name__ == "__main__":

	# from pathlib import Path
	# ROOT = Path("/mnt/shared-storage-user/zhuxiaorong/liyunhao_data/my_code/cvpr26_challenge")


	TEMPLATE_PATH = "track_1_test_demo.json"
	OUTPUT_PATH = "./track_1_test.json"


	SCORE_MODEL_FOLDERS = [
	"./result-score"
	]
	LEVEL_MODEL_FOLDERS = [
	"./result-level"
	]
	REASON_MODEL_FOLDERS = [
	"./result-reason"
	]


	build_submission(
	template_path=TEMPLATE_PATH,
	score_model_folders=SCORE_MODEL_FOLDERS,
	level_model_folders=LEVEL_MODEL_FOLDERS,
	reason_model_folders=REASON_MODEL_FOLDERS,
	output_path=OUTPUT_PATH,
	score_method="mean",
	level_method="score_mean",
	)