Upload folder using huggingface_hub

modules/color_recommender/__init__.py

@@ -0,0 +1,3 @@
+"""
+Color Recommender package initialization.
+"""

modules/color_recommender/color_framework.py

@@ -0,0 +1,224 @@
+import colour, json, random, os
+import numpy as np
+from pathlib import Path
+import sys
+
+# 获取当前文件的绝对路径
+current_file = Path(__file__).resolve()
+
+# 获取项目根目录路径 (假设config.py在根目录)
+project_root = current_file.parent.parent.parent
+
+# 将项目根目录添加到Python路径中
+if str(project_root) not in sys.path:
+    sys.path.append(str(project_root))
+
+from config import sentence_transformer_path, infographic_library_path, infographic_image_path, color_cache_path, test_c2t_root
+from llm_api import ask, ask_image
+from infographic_retrieve import InfographicRetriever
+
+ifg_retriever = InfographicRetriever(sentence_transformer_path, infographic_library_path, infographic_image_path)
+
+encoding_prompt = """You are a data visualization expert. Based on the provided JSON data description for a chart, recommend appropriate color encodings that best represent the data.
+
+Input JSON contains:
+- Chart title, description and main insight
+- Column definitions with names, importance, descriptions and roles (x, y, group, etc.)
+- Data samples
+- Chart type specification
+- Data format details
+
+Analyze the data and provide color encoding recommendations in the following JSON format:
+
+{
+  "color_encoding_options": [
+    // Array of possible encoding columns in recommended priority order
+    // Each item should be a column role from input JSON or "none"
+    // IMPORTANT: Only include "none" if using the same color would NOT result in information loss
+    // If different colors are needed to distinguish data points, "none" should not be included
+  ],
+  "recommendations": [
+    // recommended choices for each column
+    {
+      "color_encoding_option": "column_role",
+      "explanation": "Explanation of why this encoding is appropriate",
+      "color_scheme": "monochrome|dual-color|colorful",
+      "confidence": 0.8
+    }
+  ]
+}
+
+For color schemes:
+- "monochrome": Single color with variations in saturation/lightness
+- "dual-color": Two contrasting colors (e.g., blue/orange)
+- "colorful": Multiple distinct colors
+
+Consider:
+1. Semantic meaning of data (sentiments, categories, etc.)
+2. Number of distinct groups/categories
+3. Chart type and visual clarity
+
+"""
+
+append_info = """
+The JSON data is: {json_data}
+"""
+
+encode_result_case = """```json
+{
+  "color_encoding_options": [
+    "group"
+  ],
+  "recommendations": [
+    {
+      "color_encoding_option": "group",
+      "explanation": "Encoding 'Sentiment' with color is crucial for distinguishing between positive, neutral, and negative headline shares within each person's stacked bar.  Differentiation is essential to accurately perceive the proportions of each sentiment.",
+      "color_scheme": "colorful",
+      "confidence": 0.95
+    }
+  ]
+}
+```
+"""
+
+judge_prompt = '''You are an expert in color theory and visual design. You will receive the following information:
+1. A chart title: {title}
+2. A brief description of the chart’s subject: {description}
+3. A list of chart columns or data categories: {columns}
+4. A palette with following hex color: {colors}
+
+Your task is to evaluate how suitable this palette is for the chart based on emotional and thematic appropriateness. 
+For instance, if the chart addresses a serious topic, darker or more subdued tones might be more fitting; if the topic is about environmental awareness, shades of green may be preferable.
+
+Please provide:
+- An overall suitability score from 1 to 5 (with 5 indicating an excellent match and 1 indicating a poor match) for the palette.
+- Explanation for the score from any issues of emotional or thematic appropriateness.
+'''
+
+judge_return_case = '''
+
+Return your assessment in plain text, following the json format:
+{
+    "score": 4,
+    "Explanation": "The palette is well-suited for the chart's subject matter, with a range of colors that are both visually appealing and thematically appropriate. The colors are bright and engaging, which is ideal for a chart that aims to capture the reader's attention."
+}
+
+Thank you!'''
+
+
+
+cache_color_results = None
+def load_cache_results(indexes):
+    global cache_color_results
+    if cache_color_results is None:
+        with open(color_cache_path, 'r') as f:
+            cache_color_results = json.load(f)
+    return [cache_color_results[i] for i in indexes]
+
+class ColorFramework(object):
+    def __init__(self, data_info):
+        self.data_info = data_info
+        self.color_schemes = self.__get_color_encoding() # TODO: may add error handling
+
+    def __get_color_encoding(self):
+        # return dict if success, None if failed
+        prompt = encoding_prompt + append_info.format(json_data=self.data_info)
+        response = ask(prompt)
+        # response = encode_result_case
+        try:
+            response = json.loads(response)
+        except Exception as e:
+            response = response.split('```json')[1].split('```')[0]
+            response = json.loads(response)
+        print(response)
+
+        options = response['color_encoding_options']
+        recommendations = response['recommendations']
+        res = []
+        for recomd in recommendations:
+            option = recomd['color_encoding_option']
+            column_id = -2
+            if option == 'none':
+                column_id = -1
+            else:
+                for i, column in enumerate(self.data_info['columns']):
+                    if column['role'] == option:
+                        column_id = i
+                        break
+            if column_id == -2:
+                print('No column name existed')
+                return None
+            res.append({
+                'column_id': column_id,
+                'column_name': self.data_info['columns'][column_id]['name'],
+                'column_role': option,
+                'color_scheme': recomd['color_scheme'],
+                'confidence': recomd['confidence']
+            })
+        return res
+    
+    def load_scheme_list(self):
+        return self.color_schemes
+    
+    def get_infographic_palette(self, scheme_id, topk_num=20, query_threshold=0.3, valid_threshold=3.0, random_seed=-1):
+        # return list if success, None if failed
+        scheme = self.color_schemes[scheme_id]
+        search_query = f"{self.data_info['title']} {self.data_info['description']} {self.data_info['main_insight']} {self.data_info['columns'][scheme['column_id']]['name']}"
+        # 1. first retrieve by semantic
+        results = ifg_retriever.retrieve_similar_entries(search_query, top_k=topk_num)
+        max_similarity = results[0][1]
+        results_f = [r for r in results if r[1] >= query_threshold * max_similarity]
+
+        # 2. filter color mode
+        results_ids = [r[2] for r in results_f]
+        palettes = load_cache_results(results_ids)
+        pkg_palettes = zip(results_f, palettes)
+        filter_palettes = [palette for palette in pkg_palettes if palette[1]['mode'] == scheme['color_scheme']]
+        # print(len(filter_palettes))
+        # print(filter_palettes)
+        # print(results_f)
+        print('filter_palettes:', filter_palettes)
+
+        # 3. judge the palette score
+        scores = []
+        for palette in filter_palettes:
+            prompt = judge_prompt.format(title=self.data_info['title'], description=self.data_info['description'], columns=self.data_info['columns'], colors=palette[1]['main_color']) + judge_return_case
+            # print(prompt); exit()
+            response = ask(prompt)
+            # print(response)
+            try:
+                try:
+                    response = json.loads(response)
+                except Exception as e:
+                    # from IPython import embed
+                    # embed()
+                    response = response.split('```json')[1].split('```')[0]
+                    response = json.loads(response)
+            except:
+                response = {'score': 0, 'Explanation': 'Error'}
+            score = response['score']
+            scores.append(score)
+        
+        conclusion = zip(filter_palettes, scores)
+        valid_palettes = [c[0] for c in conclusion if c[1] >= valid_threshold]
+        print('find suitable results:', len(valid_palettes))
+        if len(valid_palettes) == 0:
+            print('No suitable palette found')
+            return None
+        # save all valid palettes 
+        if random_seed == -1:
+            return valid_palettes[0][1]
+        else:
+            random.seed(random_seed)
+            return random.choice(valid_palettes)[1]
+
+
+def test_ColorFramework():
+    data_file = os.path.join(test_c2t_root, "127.json")
+    data_info = json.load(open(data_file, 'r'))
+    # 1. input data_info
+    cf = ColorFramework(data_info)
+    # print(cf.data_info)
+    print(cf.get_infographic_palette(0))
+
+    # {'mode': 'colorful', 'color_list': ['#4ECBEE', '#FFB51F', '#DC776A', '#3465B1'], 'main_color': ['#4ECBEE', '#FFB51F', '#DC776A', '#3465B1'], 'bcg': '#F2EDEE', 'context_colors': ['#AFAFAF', '#86988B', '#EFDEBA', '#596873', '#2F4455'], 'similar_to_bcg': [], 'other_colors': []}

modules/color_recommender/color_index_builder.py

@@ -0,0 +1,138 @@
+import json
+import numpy as np
+import faiss
+from typing import Dict, List, Optional
+import os
+from utils.model_loader import ModelLoader
+
+class ColorIndexBuilder:
+    def __init__(self, data_path: str = "./static/color_palette.json", index_path: str = "./static/color_palette.index", embed_model_path: str = "all-MiniLM-L6-v2"):
+        # Convert relative path to absolute path
+        current_dir = os.path.dirname(os.path.abspath(__file__))
+        self.color_palette_path = data_path
+        # Use the global ModelLoader to get the model instance
+        self.model = ModelLoader.get_model(embed_model_path)
+        self.index = None
+        self.color_palettes = None
+        self.index_path = index_path
+        self.dimension = 384  # Dimension of the sentence transformer embeddings
+
+    def load_color_palettes(self) -> Dict[str, Dict]:
+        """Load color palettes from the JSON file."""
+        with open(self.color_palette_path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+
+    def create_text_for_embedding(self, palette: Dict) -> str:
+        """Create a text string from palette metadata for embedding."""
+        text_parts = []
+        
+        # Add text if available
+        if 'text' in palette:
+            text_parts.append(palette['text'])
+            
+        # Add facts if available
+        if 'facts' in palette:
+            facts = palette['facts']
+            if isinstance(facts, list):
+                text_parts.extend(facts)
+            elif isinstance(facts, str):
+                text_parts.append(facts)
+                
+        # Add columns if available
+        if 'columns' in palette:
+            columns = palette['columns']
+            if isinstance(columns, list):
+                text_parts.extend(columns)
+            elif isinstance(columns, str):
+                text_parts.append(columns)
+                
+        return ' '.join(text_parts)
+
+    def build_index(self):
+        """Build the FAISS index for color palettes."""
+        # Load color palettes
+        self.color_palettes = self.load_color_palettes()
+        
+        # Create embeddings for each palette
+        embeddings = []
+        self.palette_indices = []
+        
+        for index, palette in self.color_palettes.items():
+            text = self.create_text_for_embedding(palette)
+            print('text', text)
+            embedding = self.model.encode(text)
+            embeddings.append(embedding)
+            self.palette_indices.append(index)
+            
+        # Convert to numpy array
+        embeddings = np.array(embeddings).astype('float32')
+        
+        # Create and train the index
+        self.index = faiss.IndexFlatL2(self.dimension)
+        self.index.add(embeddings)
+
+    def find_similar_palettes(self, query_text: str, k: int = 5) -> List[Dict]:
+        """
+        Find similar color palettes based on text query.
+        
+        Args:
+            query_text: Text to search for similar palettes
+            k: Number of similar palettes to return
+            
+        Returns:
+            List of similar color palettes with their distances
+        """
+        if self.index is None:
+            self.build_index()
+            
+        # Create embedding for query
+        query_embedding = self.model.encode(query_text)
+        query_embedding = np.array([query_embedding]).astype('float32')
+        
+        # Search the index
+        distances, indices = self.index.search(query_embedding, k)
+        
+        # Return similar palettes with their distances
+        results = []
+        for i, idx in enumerate(indices[0]):
+            if idx < len(self.palette_indices):  # Ensure index is valid
+                palette_index = self.palette_indices[idx]
+                results.append({
+                    'palette': self.color_palettes[palette_index],
+                    'distance': float(distances[0][i])
+                })
+                
+        return results
+
+    def save_index(self, output_path: str):
+        """Save the FAISS index to disk."""
+        if self.index is None:
+            raise ValueError("Index has not been built yet")
+        # Convert relative path to absolute path
+        #current_dir = os.path.dirname(os.path.abspath(__file__))
+        abs_output_path = output_path
+        faiss.write_index(self.index, abs_output_path)
+        
+        # Save palette indices mapping
+        indices_path = output_path + ".indices"
+        with open(indices_path, 'w', encoding='utf-8') as f:
+            json.dump(self.palette_indices, f)
+
+    def load_index(self):
+        """Load a FAISS index from disk."""
+        # Convert relative path to absolute path
+        # current_dir = os.path.dirname(os.path.abspath(__file__))
+        index_path = self.index_path
+        self.index = faiss.read_index(index_path)
+        
+        # Load color palettes
+        self.color_palettes = self.load_color_palettes()
+        
+        # Load palette indices mapping
+        indices_path = index_path + ".indices"
+        if os.path.exists(indices_path):
+            with open(indices_path, 'r', encoding='utf-8') as f:
+                self.palette_indices = json.load(f)
+        else:
+            # For backward compatibility, create indices from keys
+            self.palette_indices = list(self.color_palettes.keys())

modules/color_recommender/color_recommender.py

@@ -0,0 +1,760 @@
+import json
+from typing import Dict, List, Optional, Union
+import pandas as pd
+import random
+import argparse
+import os
+from copy import deepcopy
+from logging import getLogger
+from modules.color_recommender.color_index_builder import ColorIndexBuilder
+
+import numpy as np
+if not hasattr(np, "asscalar"):
+    np.asscalar = lambda x: x.item()
+
+from colormath.color_objects import sRGBColor, LabColor
+from colormath.color_conversions import convert_color
+from colormath.color_diff import delta_e_cie2000
+from itertools import combinations
+
+def hex_to_lab(hex_color):
+    """HEX -> LAB"""
+    hex_color = hex_color.lstrip("#")
+    r = int(hex_color[0:2], 16)
+    g = int(hex_color[2:4], 16)
+    b = int(hex_color[4:6], 16)
+
+    rgb = sRGBColor(r, g, b, is_upscaled=True)
+    lab = convert_color(rgb, LabColor)
+    return lab
+
+def color_distance(hex1, hex2):
+    """计算两个颜色的 ΔE (CIEDE2000)"""
+    lab1 = hex_to_lab(hex1)
+    lab2 = hex_to_lab(hex2)
+    return delta_e_cie2000(lab1, lab2)
+
+def mark_color_separation(colors):
+    """
+    计算所有 mark 颜色之间的区分度
+    """
+    results = []
+    for c1, c2 in combinations(colors, 2):
+        d = color_distance(c1, c2)
+        results.append({
+            "color1": c1,
+            "color2": c2,
+            "deltaE": round(d, 2)
+        })
+    minD = 1000
+    for item in results:
+        minD = min(minD, item["deltaE"])
+    return results, minD
+
+def srgb_to_linear(c):
+    """sRGB → 线性RGB"""
+    c = c / 255.0
+    if c <= 0.04045:
+        return c / 12.92
+    else:
+        return ((c + 0.055) / 1.055) ** 2.4
+
+def relative_luminance(hex_color):
+    """计算WCAG相对亮度"""
+    r, g, b = hex_to_rgb(hex_color)
+    r_lin = srgb_to_linear(r*255)
+    g_lin = srgb_to_linear(g*255)
+    b_lin = srgb_to_linear(b*255)
+    # print((r, g, b), (r_lin, g_lin, b_lin))
+
+    return 0.2126 * r_lin + 0.7152 * g_lin + 0.0722 * b_lin
+
+def contrast_ratio(color1, color2):
+    """计算WCAG对比度"""
+    l1 = relative_luminance(color1)
+    l2 = relative_luminance(color2)
+
+    lighter = max(l1, l2)
+    darker = min(l1, l2)
+
+    # print(color1, color2, lighter, darker, (lighter + 0.05) / (darker + 0.05))
+
+    return (lighter + 0.05) / (darker + 0.05)
+
+def rgb_to_hex(r, g, b):
+    r = max(0, min(int(r), 255))
+    g = max(0, min(int(g), 255))  
+    b = max(0, min(int(b), 255))
+    return '#{:02x}{:02x}{:02x}'.format(r, g, b)
+
+import colorsys
+
+# 复用之前的函数
+def adapt_color_for_dark_background(color, brightness_adjustment=25, saturation_adjustment=15, 
+                                    preserve_grays=True, min_brightness=60, max_brightness=85,
+                                    contrast_enhancement=10):
+    """将适用于浅色背景的颜色转换为适用于深色背景的颜色"""
+    rgb = hex_to_rgb(color)
+    h, s, l = rgb_to_hsl(*rgb)
+    
+    perceived_brightness = (0.299 * rgb[0] + 0.587 * rgb[1] + 0.114 * rgb[2])
+    if preserve_grays and s < 0.1:
+        s = min(0.15, s + 0.1)  # 增加一些饱和度，但不超过15%
+        l = max(min_brightness/100, l + brightness_adjustment/100)  # 显著提高亮度
+    else:
+        # 动态调整饱和度 - 根据原始亮度调整
+        dynamic_sat_adjustment = saturation_adjustment * (1 - perceived_brightness) * 1.5
+        
+        s = max(0, min(1, s + dynamic_sat_adjustment/100))
+        dynamic_brightness_adjustment = brightness_adjustment * (1 - perceived_brightness) * 1.5
+        l = max(0, min(1, l + dynamic_brightness_adjustment/100))
+        
+        # 增强对比度
+        if l < 0.5:
+            contrast_factor = contrast_enhancement / 200  
+            l = max(0, l - contrast_factor * (0.5 - l))
+        else:
+            contrast_factor = contrast_enhancement / 200
+            l = min(1, l + contrast_factor * (l - 0.5))
+    
+    l = max(min_brightness/100, l)
+    l = min(max_brightness/100, l)
+    
+    hsl_str = f"hsl({int(h*360)}°, {int(s*100)}%, {int(l*100)}%)"
+    hex_str = hsl_to_hex(h, s, l)
+    
+    return hex_str
+def adapt_color_for_light_background(color, brightness_reduction=15, saturation_adjustment=5, 
+                                    preserve_grays=True, min_brightness=25, max_brightness=90,  # Increased max_brightness
+                                    contrast_enhancement=10):
+    """
+    将颜色优化为适用于浅色背景的颜色
+    
+    参数:
+        color (str): 颜色值，支持十六进制格式(#RGB 或 #RRGGBB)
+        brightness_reduction (int): 亮度降低百分比 (0-100)
+        saturation_adjustment (int): 饱和度调整百分比 (-100 到 100)
+        preserve_grays (bool): 是否保持灰色调的特殊处理
+        min_brightness (int): 最小亮度阈值 (0-100)
+        max_brightness (int): 最大亮度阈值 (0-100)
+        contrast_enhancement (int): 对比度增强系数 (0-100)
+        
+    返回:
+        tuple: 转换后的HSL颜色字符串和十六进制颜色字符串的元组
+    """
+    # ��十六进制颜色转换为RGB
+    rgb = hex_to_rgb(color)
+    h, s, l = rgb_to_hsl(*rgb)
+    perceived_brightness = (0.299 * rgb[0] + 0.587 * rgb[1] + 0.114 * rgb[2])
+
+    if preserve_grays and s < 0.1:
+        # 灰色调在浅色背景上需要足够深，但不要太黑
+        l = max(min_brightness/100, min(0.6, l - 0.1))  # Adjusted to make it lighter
+        s = min(0.05, s)  # 保持低饱和度
+    else:
+        if l > 0.7:
+            dynamic_brightness_reduction = brightness_reduction * (perceived_brightness)
+            l = max(min_brightness/100, l - dynamic_brightness_reduction/100)
+        
+        if l < 0.3:
+            l = min(0.5, l + 0.1)  # Adjusted to make it lighter
+
+        if s < 0.4 and l > 0.5:
+            s = min(0.6, s + saturation_adjustment/100 * 2)
+        elif s > 0.8 and l < 0.4:
+            s = max(0.6, s - saturation_adjustment/100)
+    
+    if l > 0.65:
+        contrast_factor = contrast_enhancement / 200
+        l = max(min_brightness/100, l - contrast_factor * (l - 0.5))
+    elif l < 0.45:
+        contrast_factor = contrast_enhancement / 300
+        l = min(0.6, l + contrast_factor * (0.5 - l))  # Adjusted to make it lighter
+    
+    l = max(min_brightness/100, l)
+    l = min(max_brightness/100, l)
+    
+    hsl_str = f"hsl({int(h*360)}°, {int(s*100)}%, {int(l*100)}%)"
+    hex_str = hsl_to_hex(h, s, l)
+    
+    return hex_str
+
+def convert_palette_for_light_background(palette, **kwargs):
+    """
+    转换整个调色板以适应浅色背景
+    
+    参数:
+        palette (list): 颜色列表，每个颜色为十六进制格式
+        **kwargs: 传递给adapt_color_for_light_background的选项
+        
+    返回:
+        list: 转换后的颜色列表，每个元素为(hsl, hex)的元组
+    """
+    converted_palette = []
+    
+    for color in palette:
+        converted = adapt_color_for_light_background(color, **kwargs)
+        converted_palette.append(converted)
+    
+    return converted_palette
+
+def enhance_palette_accessibility(palette, background_is_dark=True, **kwargs):
+    """
+    根据背景色增强调色板的可访问性
+    
+    参数:
+        palette (list): 颜色列表，每个颜色为十六进制格式
+        background_is_dark (bool): 背景是否为深色
+        **kwargs: 配置选项
+        
+    返回:
+        list: 转换后的颜色列表，每个元素为(hsl, hex)的元组
+    """
+    if background_is_dark:
+        return convert_palette_for_dark_background(palette, **kwargs)
+    else:
+        return convert_palette_for_light_background(palette, **kwargs)
+
+def hex_to_rgb(hex_color):
+    """将十六进制颜色转换为RGB"""
+    hex_color = hex_color.lstrip('#')
+    if len(hex_color) == 3:
+        hex_color = ''.join([c + c for c in hex_color])
+
+    r = int(hex_color[0:2], 16) / 255.0
+    g = int(hex_color[2:4], 16) / 255.0
+    b = int(hex_color[4:6], 16) / 255.0
+    
+    return (r, g, b)
+
+def rgb_to_hsl(r, g, b):
+    """将RGB转换为HSL"""
+    h, l, s = colorsys.rgb_to_hls(r, g, b)
+    return (h, s, l)
+
+def hsl_to_hex(h, s, l):
+    """将HSL转换为十六进制颜色"""
+    r, g, b = colorsys.hls_to_rgb(h, l, s)
+    
+    r = int(r * 255)
+    g = int(g * 255)
+    b = int(b * 255)
+    
+    return f"#{r:02x}{g:02x}{b:02x}"
+
+def convert_palette_for_dark_background(palette, **kwargs):
+    """
+    转换整个调色板以适应深色背景
+    
+    参数:
+        palette (list): 颜色列表，每个颜色为十六进制格式
+        **kwargs: 传递给adapt_color_for_dark_background的选项
+        
+    返回:
+        list: 转换后的颜色列表，每个元素为(hsl, hex)的元组
+    """
+    converted_palette = []
+    
+    for color in palette:
+        converted = adapt_color_for_dark_background(color, **kwargs)
+        converted_palette.append(converted)
+    
+    return converted_palette
+
+
+fixed_color_palette = [
+    ["#4269d0","#efb118","#ff725c","#6cc5b0","#3ca951","#ff8ab7","#a463f2","#97bbf5","#9c6b4e","#9498a0"],
+    ["#4e79a7","#f28e2c","#e15759","#76b7b2","#59a14f","#edc949","#af7aa1","#ff9da7","#9c755f","#bab0ab"],
+    ["#66c2a5","#fc8d62","#8da0cb","#e78ac3","#a6d854","#ffd92f","#e5c494","#b3b3b3"],
+]
+
+# (bg_hex, text_hex) pairs — text is a dark tone that harmonizes with the bg hue
+light_background_palette = [
+    # neutral / warm-white
+    ("#F5F5F0", "#2C2C28"), ("#F3F3F3", "#2A2A2A"), ("#EAEAEA", "#282828"),
+    ("#F0EEE8", "#2C2A22"), ("#F2F2E9", "#2A2A20"), ("#F3ECD9", "#332B18"),
+    ("#ECE3D9", "#32281E"), ("#EFE7D8", "#332A1C"), ("#E6E3D9", "#302A20"),
+    # warm: peach / sand / butter
+    ("#F7EDE2", "#3D2510"), ("#F5E6D3", "#3A2210"), ("#F2DFC8", "#3A2010"),
+    ("#F9F0DC", "#3A3010"), ("#F5EDD0", "#38300E"), ("#F2E8C4", "#362E0C"),
+    ("#FAF0E6", "#3C2C14"), ("#F8E8D8", "#3A2818"),
+    # warm: blush / rose
+    ("#F5E8E8", "#3D1818"), ("#F2E0E0", "#3A1616"), ("#F0D8D8", "#381414"),
+    ("#F7EAEA", "#3C1A1A"), ("#EFE0E4", "#38161C"),
+    # cool: sky / slate-blue
+    ("#E8F0F7", "#18283D"), ("#E2EBF5", "#162438"), ("#DCE8F2", "#142030"),
+    ("#E6EEF8", "#1A2A3C"), ("#E0EAF5", "#162238"), ("#D8E8F0", "#121E2E"),
+    ("#E8F4F8", "#183040"),
+    # cool: mint / sage
+    ("#E6F2EE", "#183028"), ("#DFF0EA", "#142C24"), ("#E0EDE8", "#182A22"),
+    ("#E8F5F0", "#1A3228"), ("#D8EDE6", "#102820"),
+    # cool: lavender / lilac
+    ("#EEE8F5", "#2A1840"), ("#EAE2F5", "#261638"), ("#E8E0F0", "#221430"),
+    ("#F0EAF7", "#2C1A42"), ("#ECE4F2", "#28163C"),
+    # cool: soft-teal
+    ("#E0F0F2", "#10282C"), ("#D8EEF0", "#0E2428"), ("#E4F2F4", "#122A2E"),
+    # warm-green / olive
+    ("#EAF0E0", "#222C10"), ("#E8EDD8", "#202A0E"), ("#E4EAD4", "#1E280C"),
+    ("#EDF2E2", "#242E12"),
+    # medium-toned (slightly more saturated, still light)
+    ("#DDE8F0", "#122030"), ("#D8E8E0", "#102820"), ("#E8DDF0", "#201030"),
+    ("#F0DDE8", "#301020"), ("#F0E8DD", "#302010"),
+]
+
+light_background_colors = [bg for bg, _ in light_background_palette]
+
+
+def _text_color_for_bg(bg_hex: str) -> str:
+    """Return a harmonized dark text color for the given light background."""
+    for bg, text in light_background_palette:
+        if bg.upper() == bg_hex.upper():
+            return text
+    # fallback: use luminance-derived neutral dark
+    try:
+        r, g, b = int(bg_hex[1:3], 16), int(bg_hex[3:5], 16), int(bg_hex[5:7], 16)
+        # shift hue toward darker version of the bg
+        h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255)
+        v2 = max(0.12, v - 0.68)
+        s2 = min(1.0, s * 1.2)
+        r2, g2, b2 = colorsys.hsv_to_rgb(h, s2, v2)
+        return rgb_to_hex(int(r2 * 255), int(g2 * 255), int(b2 * 255))
+    except Exception:
+        return "#2a2a2a"
+
+dark_background_colors = [
+    "#1e2130", "#1a1a2e", "#191919", "#162447", 
+    "#2d3142", "#1f2b44", "#2b2b3d", "#282828", "#243447", 
+    "#373e57", "#343434", "#203a43", "#1c2541", "#2c2c34", 
+    "#2e2e2e", "#254362", "#352f44", "#204051", "#303030"
+]
+
+class ColorRecommender:
+    def __init__(self, embed_model_path: str = "all-MiniLM-L6-v2", data_path: str = None, index_path: str = None):
+        self.color_schemes = {}
+        self.index_builder = ColorIndexBuilder(embed_model_path=embed_model_path, data_path=data_path, index_path=index_path)
+        self.index_builder.load_index()
+        black = (21, 21, 21)
+        white = (240, 240, 240)
+        gray1 = (75, 75, 75)
+        gray2 = (150, 150, 150)
+        gray3 = (200, 200, 200)
+        self.basic_colors = [black, white, gray1, gray2, gray3]
+        self.basic_colors_hex = [rgb_to_hex(*color) for color in self.basic_colors]
+
+    def create_query_text(self, input_data: Dict) -> str:
+        """Create a text query from input data for finding similar color palettes."""
+        data_dict = input_data.get("data", {})
+        columns = data_dict.get("columns", [])
+        metadata = input_data.get("metadata", {})
+        
+        text_parts = []
+        
+        # Add column information
+        column_texts = []
+        for col in columns:
+            col_text = f"{col['name']} ({col['data_type']})"
+            if 'description' in col:
+                col_text += f": {col['description']}"
+            column_texts.append(col_text)
+        text_parts.append(" ".join(column_texts))
+        
+        # Add metadata
+        if 'title' in metadata:
+            text_parts.append(metadata['title'])
+        if 'description' in metadata:
+            text_parts.append(metadata['description'])
+        if 'main_insight' in metadata:
+            text_parts.append(metadata['main_insight'])
+            
+        return " ".join(text_parts)
+
+    def determine_by_group(self, columns: List[Dict], combination: str) -> Optional[str]:
+        """
+        Determine which column to use for grouping based on the combination type.
+        
+        Args:
+            columns: List of column dictionaries containing name and data_type
+            combination: Type of combination (e.g., "categorical + numerical")
+            
+        Returns:
+            The column name to use for grouping, or None if no grouping should be used
+        """
+        column_names = [col["name"] for col in columns]
+        if combination == "categorical + numerical" or combination == "categorical + numerical + numerical":
+            return column_names[0]
+        elif combination == "categorical + numerical + categorical":
+            return column_names[2]
+        elif combination == "temporal + numerical + categorical":
+            return column_names[2]
+        elif combination == "temporal + numerical":
+            return column_names[0]
+        elif combination == "categorical + numerical + temporal":
+            return column_names[0]
+        else:
+            ret = None
+            for col in columns:
+                if col["data_type"] == "categorical" or col["data_type"] == "temporal":
+                    ret = col["name"]
+            return ret
+
+    def should_color_by_group(self, by_group: str, data: pd.DataFrame, columns: List[Dict]) -> bool:
+        """
+        Determine whether to use color grouping based on the column's unique values and type.
+        
+        Args:
+            by_group: Column name to check
+            data: DataFrame containing the data
+            columns: List of column dictionaries containing name and data_type
+            
+        Returns:
+            True if color grouping should be used, False otherwise
+        """
+        if by_group is None:
+            return False
+            
+        unique_values = data[by_group].nunique()
+        column_type = next((col["data_type"] for col in columns if col["name"] == by_group), "")
+        
+        if column_type == "temporal" and unique_values >= 5:
+            return False
+        elif column_type == "categorical" and unique_values >= 8:
+            return False
+            
+        return True
+
+    def get_required_color_num(self, by_group: Optional[str], data: pd.DataFrame) -> int:
+        """
+        Calculate the number of colors required based on grouping.
+        
+        Args:
+            by_group: Column name used for grouping
+            data: DataFrame containing the data
+            
+        Returns:
+            Number of colors required
+        """
+        if by_group is None:
+            return 1
+        return data[by_group].nunique()
+
+    def select_suitable_palettes(self, similar_palettes: List[Dict], required_color_num: int) -> Dict:
+        """
+        Select a suitable palette from similar palettes based on color count requirements.
+        
+        Args:
+            similar_palettes: List of similar palettes with their distances
+            required_color_num: Number of colors required
+            
+        Returns:
+            Selected palette dictionary
+        """
+        # Filter palettes that have enough colors
+        suitable_palettes = [
+            p for p in similar_palettes 
+            if 'main_color' in p['palette'] and len(p['palette']['main_color']) >= required_color_num + 1 and len(p['palette']['main_color']) <= required_color_num + 3
+        ]
+        
+        if len(suitable_palettes) == 0:
+            # If no palette has enough colors, use the most similar one
+            # return random.choice(similar_palettes[:3])['palette']
+            return [similar_palette['palette'] for similar_palette in similar_palettes[:3]]
+            
+        # TODO: Implement custom selection logic based on requirements
+        # return random.choice(suitable_palettes[:3])['palette']
+        return [suitable_palette['palette'] for suitable_palette in suitable_palettes[:3]]
+
+    def recommend_colors(self, input_data: Dict) -> Dict:
+        """
+        Recommend colors based on the input data.
+        
+        Args:
+            input_data: Dictionary containing the input data
+            
+        Returns:
+            Dictionary containing the recommended color scheme
+        """
+        # Extract necessary information from input_data
+        data_dict = input_data.get("data", {})
+        data = pd.DataFrame(data_dict.get("data", []))
+        columns = data_dict.get("columns", [])
+        combination = data_dict.get("type_combination", "")
+        
+        # Step 1: Determine by_group
+        by_group = self.determine_by_group(columns, combination)
+        
+        # Step 2: Check if we should color by group
+        should_group = self.should_color_by_group(by_group, data, columns)
+        if not should_group:
+            by_group = None
+            
+        # Step 3: Get required number of colors
+        required_color_num = self.get_required_color_num(by_group, data)
+        
+        # Step 4: Find similar color palettes
+        query_text = self.create_query_text(input_data)
+        similar_palettes = self.index_builder.find_similar_palettes(query_text, k=25)
+        # Step 5: Select a suitable palette
+        selected_palettes = self.select_suitable_palettes(similar_palettes, required_color_num)
+        if not selected_palettes or len(selected_palettes) == 0:
+            selected_palettes = [None]
+
+        color_schemes = []
+        for selected_palette in selected_palettes:
+            if not selected_palette or len(selected_palette["main_color"]) <= required_color_num:
+                scheme = fixed_color_palette[random.randint(0, len(fixed_color_palette) - 1)]
+                new_selected_palette = {
+                    "mode": "monochrome",
+                    "color_list": scheme,
+                    "main_color": scheme,
+                    "num_of_colors": len(scheme),
+                    "bcg": selected_palette["bcg"] if selected_palette else random.choice(light_background_colors),
+                    "context_colors": selected_palette["context_colors"] if selected_palette else [],
+                    "similar_to_bcg": selected_palette["similar_to_bcg"] if selected_palette else 0,
+                }
+                selected_palette = new_selected_palette
+
+            # Step 6: Create the color scheme
+            chosen_bg = random.choice(light_background_colors)
+            color_scheme = {
+                "field": {},
+                "other": {
+                    "primary": None,
+                },
+                "available_colors": [],
+                "background_color": chosen_bg,
+                "text_color": _text_color_for_bg(chosen_bg)
+            }
+            
+            colors = selected_palette["main_color"] + selected_palette["context_colors"]
+            if by_group:
+                unique_values = data[by_group].unique()
+                first_numerical_value = None
+                for col in columns:
+                    if col["data_type"] == "numerical":
+                        first_numerical_value = col["name"]
+                        break
+                unique_values = [(value, data[first_numerical_value][data[by_group] == value].mean()) for value in unique_values]
+                unique_values.sort(key=lambda x: -x[1])
+                unique_values = [value[0] for value in unique_values]
+                color_scheme["other"]["primary"] = selected_palette["main_color"][0]
+                for i, value in enumerate(unique_values):
+                    color_scheme["field"][str(value)] = selected_palette["main_color"][i]
+                if required_color_num < len(colors):
+                    color_scheme["other"]["secondary"] = colors[required_color_num]
+                    for i in range(required_color_num + 1, len(colors)):
+                        color_scheme["available_colors"].append(colors[i])
+            else:
+                required_color_num = 1
+                color_scheme["other"]["primary"] = selected_palette["main_color"][0]
+                if required_color_num < len(colors):
+                    color_scheme["other"]["secondary"] = colors[required_color_num]
+                    for i in range(required_color_num + 1, len(colors)):
+                        color_scheme["available_colors"].append(colors[i])
+            
+            color_schemes.append(color_scheme)
+        
+        return color_schemes
+
+def process(input: str, output: str, embed_model_path: str = "all-MiniLM-L6-v2", base_url: str = None, api_key: str = None, data_path: str = None, index_path: str = None) -> bool:
+    """
+    Pipeline入口函数，处理单个文件的颜色推荐
+    
+    Args:
+        input_path: 输入JSON文件路径
+        output_path: 输出JSON文件路径
+        embed_model_path: 嵌入模型路径
+    """
+    print(f"Processing {input} to {output}")
+    try:
+        # 读取输入文件
+        with open(input, "r", encoding="utf-8") as f:
+            data = json.load(f)
+            
+        # 预处理数据，确保类型正确
+        processed_data = preprocess_data(data)
+        
+        # 生成颜色推荐
+        recommender = ColorRecommender(embed_model_path=embed_model_path, data_path=data_path, index_path=index_path)
+        color_results = recommender.recommend_colors(processed_data)
+        
+        color_results2 = []
+        for color_result in color_results:
+            lighter_color_result = deepcopy(color_result)
+            for k, v in lighter_color_result.items():
+                if isinstance(v, list):
+                    lighter_color_result[k] = convert_palette_for_light_background(v)
+                elif isinstance(v, dict):
+                    for k2, v2 in v.items():
+                        lighter_color_result[k][k2] = convert_palette_for_light_background([v2])[0]
+                        if "other" in k2.lower() and len(k2) <= 7:
+                            lighter_color_result[k][k2] = "#6f6f6f"
+                elif isinstance(v, str) and k != "background_color":
+                    lighter_color_result[k] = convert_palette_for_light_background([v])[0]
+                # elif k == "background_color":
+                #     darker_color_result[k] = random.choice(light_background_colors)
+
+            darker_color_result = deepcopy(color_result)
+            for k, v in darker_color_result.items():
+                if isinstance(v, list):
+                    darker_color_result[k] = convert_palette_for_dark_background(v)
+                elif isinstance(v, dict):
+                    for k2, v2 in v.items():
+                        darker_color_result[k][k2] = convert_palette_for_dark_background([v2])[0]
+                        if "other" in k2.lower() and len(k2) <= 7:
+                            darker_color_result[k][k2] = "#8f8f8f"
+                elif isinstance(v, str) and k != "background_color":
+                    darker_color_result[k] = convert_palette_for_dark_background([v])[0]
+                elif k == "background_color":
+                    darker_color_result[k] = random.choice(dark_background_colors)
+            
+            def checkWCAG(color_result):
+                minWCAG = 100
+                for key in color_result["field"]:
+                    minWCAG = min(minWCAG, contrast_ratio(color_result["field"][key], color_result["background_color"]))
+                for key in color_result["other"]:
+                    minWCAG = min(minWCAG, contrast_ratio(color_result["other"][key], color_result["background_color"]))
+                return minWCAG
+            
+            # 希望WCAG大于等于1.5，否则重新选背景色
+            best_minWCAG = checkWCAG(lighter_color_result)
+            # print("light WCAG", best_minWCAG)
+            if best_minWCAG < 1.5:
+                best_bcg = lighter_color_result["background_color"]
+                for color in random.sample(light_background_colors, len(light_background_colors)):
+                    lighter_color_result["background_color"] = color
+                    minWCAG = checkWCAG(lighter_color_result)
+                    # print("now light WCAG", minWCAG)
+                    if minWCAG > best_minWCAG:
+                        best_bcg = color
+                        best_minWCAG = minWCAG
+                    if best_minWCAG >= 1.5:
+                        break
+                lighter_color_result["background_color"] = best_bcg
+            # print("light WCAG", best_minWCAG)
+
+            # 希望WCAG大于等于3，否则重新选背景色
+            best_minWCAG = checkWCAG(darker_color_result)
+            # print("dark WCAG", best_minWCAG)
+            if best_minWCAG < 3:
+                best_bcg = darker_color_result["background_color"]
+                for color in random.sample(dark_background_colors, len(dark_background_colors)):
+                    darker_color_result["background_color"] = color
+                    minWCAG = checkWCAG(darker_color_result)
+                    if minWCAG > best_minWCAG:
+                        best_bcg = color
+                        best_minWCAG = minWCAG
+                    if best_minWCAG >= 3:
+                        break
+                darker_color_result["background_color"] = best_bcg
+            # print("dark WCAG", best_minWCAG)
+
+            color_results2.append((lighter_color_result, darker_color_result))
+        
+        def checkSeparation(color_result):
+            color_list = [color_result["background_color"]]
+            for key in color_result["field"]:
+                color_list.append(color_result["field"][key])
+            for key in color_result["other"]:
+                color_list.append(color_result["other"][key])
+            _, minD = mark_color_separation(list(set(color_list)))
+            return minD
+
+        random.shuffle(color_results2)
+
+        sep_thres = 20
+        # 过滤掉颜色区分度低(minD < sep_thres)的方案, 添加颜色方案到数据中
+        best_idx = 0
+        best_minD = checkSeparation(color_results2[best_idx][0])
+        # print("light minD", best_minD)
+        if best_minD < sep_thres:
+            for idx in range(1, len(color_results2)):
+                minD = checkSeparation(color_results2[idx][0])
+                if minD > best_minD:
+                    best_minD = minD
+                    best_idx = idx
+                if best_minD >= sep_thres:
+                    break
+        # print("light minD", best_minD)
+        processed_data["colors"] = color_results2[best_idx][0]
+
+        best_idx = 0
+        best_minD = checkSeparation(color_results2[best_idx][1])
+        # print("dart minD", best_minD)
+        if best_minD < sep_thres:
+            for idx in range(1, len(color_results2)):
+                minD = checkSeparation(color_results2[idx][1])
+                if minD > best_minD:
+                    best_minD = minD
+                    best_idx = idx
+                if best_minD >= sep_thres:
+                    break
+        # print("dart minD", best_minD)
+        processed_data["colors_dark"] = color_results2[best_idx][1]
+        
+        # 保存结果
+        with open(output, "w", encoding="utf-8") as f:
+            json.dump(processed_data, f, indent=2, ensure_ascii=False)
+            
+        return True
+            
+    except Exception as e:
+        print(f"Processing {input} failed: {str(e)}")
+        return False
+
+def preprocess_data(data: Dict) -> Dict:
+    """
+    预处理数据，处理类型转换问题
+    """
+    try:
+        # 深拷贝避免修改原始数据
+        processed = data.copy()
+        
+        # 确保data字段存在且格式正确
+        if "data" in processed and isinstance(processed["data"], dict):
+            # 处理数据部分
+            if "data" in processed["data"]:
+                rows = processed["data"]["data"]
+                if isinstance(rows, list):
+                    # 处理每一行数据
+                    for i, row in enumerate(rows):
+                        if isinstance(row, dict):
+                            # 尝试将数值字符串转换为数值类型
+                            for key, value in row.items():
+                                if isinstance(value, str):
+                                    try:
+                                        # 尝试转换为数值
+                                        if '.' in value:
+                                            row[key] = float(value)
+                                        else:
+                                            row[key] = int(value)
+                                    except (ValueError, TypeError):
+                                        # 如果转换失败，保持原始值
+                                        pass
+                                elif value is None:
+                                    # 将None替换为0或其他适当的默认值
+                                    row[key] = 0
+        
+        return processed
+        
+    except Exception as e:
+        logger.error(f"数据预处理失败: {str(e)}")
+        raise
+
+def main():
+    parser = argparse.ArgumentParser(description="Color Recommender")
+    parser.add_argument("--input", type=str, required=True, help="Input JSON file path")
+    parser.add_argument("--output", type=str, required=True, help="Output JSON file path")
+    parser.add_argument("--embed_model_path", type=str, default="all-MiniLM-L6-v2", 
+                        help="Path to the embedding model")
+    args = parser.parse_args()
+
+    success = process(input=args.input, output=args.output, 
+                     embed_model_path=args.embed_model_path)
+
+    if success:
+        print("Processing json successed.")
+    else:
+        print("Processing json failed.")
+
+if __name__ == "__main__":
+    main() 

modules/color_recommender/color_recommender_to_debug.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+颜色推荐模块 (color_recommender)
+基于输入数据特征，自动推荐颜色
+"""
+
+import json
+import logging
+import argparse
+from typing import Dict, List, Any, Tuple
+from color_framework import ColorFramework
+from color_template import ColorDesign
+
+# 配置日志
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+
+def process(input: str, output: str) -> bool:
+    """
+    处理输入数据并生成颜色推荐
+    
+    Args:
+        input: 输入JSON文件路径
+        output: 输出JSON文件路径
+        
+    Returns:
+        处理成功返回True，否则返回False
+    """
+    try:
+        # 读取输入数据
+        logger.info(f"读取输入文件: {input}")
+        with open(input, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        
+        # 抽取需要使用的数据条目
+        # logger.info("抽取颜色需要使用的数据条目")
+        assert "metadata" in data, "数据格式错误，缺少'metadata'字段"
+        assert "data" in data or "columns" not in data["data"], "数据格式错误，缺少'data.columns'字段"
+        assert "datafacts" in data, "数据格式错误，缺少'datafacts'字段"
+        data_info = {
+            "title": data["metadata"]["title"],
+            "description": data["metadata"]["description"],
+            "main_insight": data["metadata"]["main_insight"],
+            "columns": data["data"]["columns"],
+            "data_facts": data["datafacts"]
+        }
+        
+        # 生成颜色模式推荐
+        # logger.info("生成颜色模式推荐")
+        colorframework = ColorFramework(data_info)
+        schemes = colorframework.load_scheme_list()
+        logger.info(f"获取颜色模式共{len(schemes)}种, 依次为: {schemes}")
+        
+        # 默认选择第一种模式配置调色盘
+        # TODO: 【存在多种可选模式】选择最合适的颜色模式，或者增加随机性得到不同的颜色模式
+        palette = colorframework.get_infographic_palette(0)
+        logger.info(f"默认选择第一种模式配置调色盘: {palette}")
+
+        # 生成颜色模版，根据颜色模版配置颜色
+        color_design = ColorDesign(palette, lighter="high") # lighter: "high","low"
+        res = {}
+        if len(palette["main_color"]) == 1: # 仅有一个主色
+            res["other"] = {
+                "primary": color_design.main_color_hex[0],
+            }
+        elif len(palette["main_color"]) == 2: # 有两个主色
+            res["other"] = {
+                "primary": color_design.main_color_hex[0],
+                "secondary": color_design.main_color_hex[1],
+            }
+        else:
+            res["other"] = {}
+
+        use_group = False
+        group_col = None
+        x_col = None
+        for column in data_info["columns"]:
+            if column["role"] == "group":
+                use_group = True
+                group_col = column["name"]
+            if column["role"] == "x":
+                x_col = column["name"]
+        # TODO：【存在多种可选颜色参数】选择最合适的颜色参数，或者增加随机性得到不同的颜色模版，例如以下给出的各种seed
+        # get_color: seed_mark（mark颜色模式）seed_text（文本颜色模式）
+        #            seed_color（gourp时不同颜色模式）, seed_context_color（不同的上下文颜色）, seed_middle_color （不同的其他中间色）
+
+        if not use_group:
+            unique_x_values = list(set(item[x_col] for item in data["data"]))
+            print("unique_x_values: ", unique_x_values)
+            mark_colors = color_design.get_color("marks", len(unique_x_values), group=1, seed_mark=0) 
+            color_list = mark_colors["group1"]
+            # TODO: 【根据语义等其他特征调整颜色分配】目前直接assign
+            color_assign = {}
+            for i, item in enumerate(unique_x_values):
+                color_assign[item] = color_list[i]
+            res["field"] = color_assign
+        else:
+            groups = {}
+            group_idx = 0
+            group_list = []
+            data_group = []
+            for item in data["data"]:
+                if item[group_col] not in groups:
+                    groups[item[group_col]] = group_idx
+                    group_idx += 1
+                    group_list.append(item[group_col])
+                    data_group.append([])
+                data_group[groups[item[group_col]]].append(item)
+            # TODO: 目前要求每个group下的data数量一样，后续可以考虑不一样的情况
+            assert len(set([len(d) for d in data_group])) == 1, "每个group下的data数量不一样"
+            mark_colors = color_design.get_color("marks", len(data_group[0]), group=len(data_group), seed_mark=1)
+            # TODO: 【根据语义等其他特征调整颜色分配】目前根���每个group直接assign
+            color_assign = {}
+            for i, group in enumerate(data_group):
+                color_list = mark_colors[f"group{i+1}"]
+                for j, item in enumerate(group):
+                    color_assign[item[x_col]] = color_list[j]
+            res["field"] = color_assign
+
+        res["background_color"] = color_design.get_color("background")[0]
+        text_colors = color_design.get_color("text", seed_text=1)
+        res["text_color"] = text_colors["annotation"][0]
+        res["title_color"] = text_colors["title"]
+        res["caption_color"] = text_colors["caption"]
+
+        used_colors = []
+        for key in res["field"]:
+            used_colors.append(res["field"][key])
+        used_colors.append(res["background_color"])
+        used_colors.append(res["text_color"])
+        res["available_colors"] = color_design.get_emphasis_colors(used_colors) 
+        # TODO: 【available颜色计算】目前仅考虑了颜色距离，可以进一步筛选可用的其他颜色
+
+        # TODO: 【考虑data fact增加可以使用的颜色】
+        data["color"] = res
+        
+        # 写入输出文件
+        logger.info(f"写入输出文件: {output}")
+        with open(output, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        
+        logger.info("颜色推荐完成")
+        return True
+        
+    except Exception as e:
+        logger.error(f"处理失败: {str(e)}")
+        return False
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="ChartPipeline - 颜色推荐模块")
+    parser.add_argument("--input", required=True, help="输入JSON文件路径")
+    parser.add_argument("--output", required=True, help="输出JSON文件路径")
+    
+    args = parser.parse_args()
+    
+    process(input=args.input, output=args.output) 

modules/color_recommender/color_static.py

@@ -0,0 +1,76 @@
+import colour
+import numpy as np
+import random
+
+def palette_iteration(lch_palette):
+    # rank the colors in the palette by hue
+    # linearly interpolate between colors in the palette
+    lchs = lch_palette[:]
+    lchs.sort(key=lambda x: x[2])
+    lchs = lchs + [lchs[0]]
+    res = []
+    for i in range(len(lchs) - 1):
+        lch1 = lchs[i]
+        lch2 = lchs[i + 1]
+        lch_inter = [(lch1[j] + lch2[j]) / 2 for j in range(3)]
+        res.append(lch1)
+        res.append(lch_inter)
+    return res
+
+from matplotlib import colors
+import matplotlib.pyplot as plt
+import numpy as np
+
+class StaticPalettes:
+    def __init__(self):
+        self.qualitative_colormaps = ['Pastel1', 'Pastel2', 'Paired', 'Accent', 
+                                    'Dark2', 'Set1', 'Set2', 'Set3', 'tab10', 
+                                    'tab20', 'tab20b', 'tab20c']
+    
+    def get_colors_from_cmap(self, cmap_name, n_colors):
+        if cmap_name not in self.qualitative_colormaps:
+            raise ValueError(f"Uncertain palette name: {cmap_name}")
+            
+        cmap = plt.get_cmap(cmap_name)
+        colors_rgb = cmap(np.linspace(0, 1, n_colors))
+        hex_colors = [colors.rgb2hex(rgb[:3]) for rgb in colors_rgb]
+        return hex_colors
+    
+    def get_named_color(self, color_name):
+        try:
+            rgb = colors.to_rgb(color_name)
+            return colors.rgb2hex(rgb)
+        except ValueError:
+            raise ValueError(f"Uncertain color name: {color_name}")
+    
+    def list_available_colormaps(self):
+        return self.qualitative_colormaps
+
+    def get_colors(self, n_colors):
+        # random select a palette
+        color_map = random.choice(self.qualitative_colormaps)
+        return self.get_colors_from_cmap(color_map, n_colors)
+    
+class EmphasisColors:
+    def __init__(self):
+        self.colors = []
+        # 默认添加几种强调色
+        self.colors.append("#FF0000")
+        self.colors.append("#00FF00")
+        self.colors.append("#0000FF")
+        self.colors.append("#FFFF00")
+        self.colors.append("#FF00FF")
+        self.colors.append("#00FFFF")
+        
+    def add_color(self, color):
+        self.colors.append(color)
+    
+    def get_color(self):
+        # 随机返回一个颜色
+        return random.choice(self.colors)
+
+if __name__ == "__main__":
+    sp = StaticPalettes()
+    print(sp.get_colors_from_cmap('tab10', 10))
+    print(sp.get_named_color('red'))
+    print(sp.list_available_colormaps())

modules/color_recommender/color_template.py

@@ -0,0 +1,648 @@
+import random
+import colour
+import numpy as np
+from color_static import palette_iteration, StaticPalettes
+
+static_palettes = StaticPalettes()
+
+def rgb_to_hex(r, g, b):
+    r = max(0, min(int(r), 255))
+    g = max(0, min(int(g), 255))  
+    b = max(0, min(int(b), 255))
+    return '#{:02x}{:02x}{:02x}'.format(r, g, b)
+
+def hex_to_rgb(hex):
+    hex = hex.lstrip('#')
+    return tuple(int(hex[i:i+2], 16) for i in (0, 2, 4))
+
+def random_rgb():
+    r = random.randint(0, 255)
+    g = random.randint(0, 255)
+    b = random.randint(0, 255)
+    return (r, g, b)
+
+def random_hex():
+    r, g, b = random_rgb()
+    return rgb_to_hex(r, g, b)
+
+def ciede2000(rgb1, rgb2):
+    # 转换RGB为Lab
+    rgb1_arr = np.array(rgb1) / 255.0
+    rgb2_arr = np.array(rgb2) / 255.0
+    
+    lab1 = colour.XYZ_to_Lab(colour.sRGB_to_XYZ(rgb1_arr))
+    lab2 = colour.XYZ_to_Lab(colour.sRGB_to_XYZ(rgb2_arr))
+    
+    # 计算CIEDE2000色差
+    delta_E = colour.delta_E(lab1, lab2, method='CIE 2000')
+    return delta_E
+
+def rgb_to_hcl(r, g, b):
+    """
+    LCh (L: 0-100, C: 0-100+, h: 0-360)
+    """
+    rgb = np.array([r, g, b]) / 255.0
+    xyz = colour.sRGB_to_XYZ(rgb)
+    lab = colour.XYZ_to_Lab(xyz)
+    lch = colour.Lab_to_LCHab(lab)
+    return lch
+
+def norm255rgb(rgb):
+    return [int(max(min(x*255, 255),0)) for x in rgb]
+
+def extend_color_in_l(rgb, number=5):
+    lch = rgb_to_hcl(*rgb)
+    res = []
+    # for l in range(45, 95, 10):
+    h = lch[2]
+    start_l = 45 if h < 85 and h > 114 else 75
+    end_l = 95
+    delta = (end_l - start_l) / number
+    for i in range(number):
+        l = start_l + i * delta
+        lch[0] = l
+        lab = colour.LCHab_to_Lab(lch)
+        xyz = colour.Lab_to_XYZ(lab)
+        rgb = colour.XYZ_to_sRGB(xyz)
+        res.append(norm255rgb(rgb))
+    res = [rgb_to_hex(*rgb) for rgb in res]
+    return res
+
+def extend_color_in_c(rgb, number=5):
+    lch = rgb_to_hcl(*rgb)
+    res = []
+    delta = (85 - 35) / number
+    for i in range(number):
+        c = 35 + i * delta
+        lch[1] = c
+        lab = colour.LCHab_to_Lab(lch)
+        xyz = colour.Lab_to_XYZ(lab)
+        rgb = colour.XYZ_to_sRGB(xyz)
+        res.append(norm255rgb(rgb))
+    res = [rgb_to_hex(*rgb) for rgb in res]
+    return res
+
+def delta_h(h1, h2):
+    dh = (h1-h2) % 360
+    if dh < 0:
+        dh += 360
+    if dh > 180:
+        dh = 360 - dh
+    return dh
+
+def lighter_color(rgb):
+    lch = rgb_to_hcl(*rgb)
+    if lch[0] < 80:
+        lch[0] += 15
+    if lch[1] < 70:
+        lch[1] += 10
+    lab = colour.LCHab_to_Lab(lch)
+    xyz = colour.Lab_to_XYZ(lab)
+    rgb = colour.XYZ_to_sRGB(xyz)
+    rgb = norm255rgb(rgb)
+    return rgb
+
+def iterate_rgb_palette(rgb_palette):
+    lch_palette = [rgb_to_hcl(*rgb) for rgb in rgb_palette]
+    lch_palette = palette_iteration(lch_palette)
+    rgb_palette = [norm255rgb(colour.XYZ_to_sRGB(colour.Lab_to_XYZ(colour.LCHab_to_Lab(lch)))) for lch in lch_palette]
+    return rgb_palette
+
+def emphasis_color(used_colors, palette):
+    # find the color with the largest distance to used colors in the palette
+    max_dist = 0
+    res = []
+    for color in palette:
+        min_dist = min([ciede2000(color, used_color) for used_color in used_colors])
+        if min_dist > 10:
+            res.append(rgb_to_hex(*color))
+    return res
+
+def check_in_palette(color, palette):
+    for p in palette:
+        if ciede2000(color, p) < 10:
+            return True
+    return False
+    
+text_types = ['title', 'caption']
+class ColorDesign:
+    def __init__(self, image_palette, lighter='high', seed_order = 0):
+        self.pool = image_palette
+        mode = image_palette['mode']
+        self.mode = mode
+        self.rgb_pool = [hex_to_rgb(color) for color in self.pool['main_color']]
+        if lighter == 'high':
+            self.rgb_pool_cp = [lighter_color(rgb) for rgb in self.rgb_pool]
+            self.rgb_pool = self.rgb_pool_cp
+        seed_order = seed_order % len(self.rgb_pool)
+        self.rgb_pool = self.rgb_pool[seed_order:] + self.rgb_pool[:seed_order]
+        self.rgb_pool_hex = [rgb_to_hex(*rgb) for rgb in self.rgb_pool]
+        self.main_color_hex = self.rgb_pool_hex
+        self.middle_color = None
+
+        black = (0, 0, 0)
+        white = (255, 255, 255)
+        gray1 = (75, 75, 75)
+        gray2 = (150, 150, 150)
+        gray3 = (200, 200, 200)
+        self.rgb_pool.sort(key=lambda x: ciede2000(x, black))
+
+        bcg = image_palette['bcg']
+        bcg_rgb = hex_to_rgb(bcg)
+        self.main_color = self.rgb_pool
+        self.bcg_color = bcg_rgb
+        dist_color_2_black = ciede2000(self.bcg_color, black)
+        dist_color_2_white = ciede2000(self.bcg_color, white)
+        if dist_color_2_black > dist_color_2_white:
+            self.lightness = 'light'
+        else:
+            self.lightness = 'dark'
+
+        if mode == 'colorful':
+            iter_rgb2 = iterate_rgb_palette(self.rgb_pool)
+            iter_rgb3 = iterate_rgb_palette(iter_rgb2)
+            self.rgb_pool_hex_2 = [rgb_to_hex(*rgb) for rgb in iter_rgb2]
+            self.rgb_pool_hex_3 = [rgb_to_hex(*rgb) for rgb in iter_rgb3]
+
+        self.basic_colors = [black, white, gray1, gray2, gray3]
+        self.basic_colors_hex = [rgb_to_hex(*color) for color in self.basic_colors]
+        
+
+    def get_emphasis_colors(self, used_colors):
+        if isinstance(used_colors[0], str):
+            used_colors = [hex_to_rgb(color) for color in used_colors]
+        hex_pool = self.pool['other_colors'] + self.pool['main_color']
+        if self.mode == 'colorful':
+            hex_pool += self.rgb_pool_hex_2 + self.rgb_pool_hex_3
+        palette = [hex_to_rgb(color) for color in hex_pool]
+        used_colors.append(self.bcg_color)
+        return emphasis_color(used_colors, palette)
+    
+    def get_reverse_color(self, color, seed = 0):
+        bcg_color_hex = rgb_to_hex(*self.bcg_color)
+        if color == bcg_color_hex:
+            return self.get_color('text', seed_text = seed, reverse = True)['title'][0]
+        else:
+            if self.lightness == 'dark':
+                if seed % 2 == 0:
+                    return self.basic_colors_hex[1]
+                return bcg_color_hex
+            else:
+                if seed % 2 == 0:
+                    return self.basic_colors_hex[0]
+                return bcg_color_hex
+
+    def get_color(self, type, number = 1, group = 1, \
+                  seed_color = 0, seed_context_color = 0, seed_middle_color = 0, \
+                  seed_text = 0, seed_mark = 0, seed_axis = 0):
+        if type == 'background':
+            return [self.pool['bcg']]
+
+        if self.mode == 'monochrome': 
+            main_color = self.main_color[0]
+            if group == 1:
+                if number < 6:
+                    extend_colors1 = extend_color_in_l(main_color)
+                    extend_colors2 = extend_color_in_c(main_color)
+                else:
+                    extend_colors1 = extend_color_in_l(main_color, number)
+                    extend_colors2 = extend_color_in_c(main_color, number)
+            else:
+                if group < 6:
+                    extend_colors1 = extend_color_in_l(main_color)
+                    extend_colors2 = extend_color_in_c(main_color)
+                else:
+                    extend_colors1 = extend_color_in_l(main_color, group)
+                    extend_colors2 = extend_color_in_c(main_color, group)
+
+            main_color_hex = rgb_to_hex(*main_color)
+            if self.lightness == 'dark':
+                other_color = self.basic_colors_hex[1]
+                gray_color = self.basic_colors_hex[4]
+            else:
+                other_color = self.basic_colors_hex[0]
+                gray_color = self.basic_colors_hex[2]
+            if seed_context_color > 0 and len(self.pool['context_color']) > 0:
+                choice = seed_context_color % len(self.pool['context_color'])
+                gray_color = self.pool['context_color'][choice]
+
+            if type == 'text':
+                seed = seed_text % 4
+                if seed == 0: # all same color
+                    return {
+                        'title': [main_color_hex],
+                        'caption': [main_color_hex],
+                        'annotation': [main_color_hex],
+                    }
+                if seed == 1: # all black/white
+                    return {
+                        'title': [other_color],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                if seed == 2: # extend in lightness
+                    if self.lightness == 'dark':
+                        res = extend_colors1[-3:]
+                    else:
+                        res = extend_colors1[:3]
+                    return {
+                        'title': res[:1],
+                        'caption': res[1:2],
+                        'annotation': res[2:3],
+                    }
+                if seed == 3: # main color + other black/white
+                    return {
+                        'title': [main_color_hex],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                    
+            if type == 'marks':
+                if group == 1:
+                    seed = seed_mark % 6
+                    if seed == 0:
+                        return {
+                            'group1': [main_color_hex for _ in range(number)],
+                        }
+                    if seed == 1:
+                        return {
+                            'group1': extend_colors1[:number],
+                        }
+                    if seed == 2:
+                        return {
+                            'group1': extend_colors1[-number:],
+                        }
+                    if seed == 3:
+                        return {
+                            'group1': extend_colors2[:number],
+                        }   
+                    if seed == 4:
+                        return {
+                            'group1': extend_colors2[-number:],
+                        }
+                    if seed == 5:
+                        return {
+                            'group1': [gray_color for _ in range(number)],
+                        }
+                else:
+                    seed = seed_mark % 4
+                    group_colors = []
+                    if seed == 0:
+                        group_colors = extend_colors1[:group]
+                    if seed == 1:
+                        group_colors = extend_colors1[-group:]
+                    if seed == 2:
+                        group_colors = extend_colors2[:group]
+                    if seed == 3:
+                        group_colors = extend_colors2[-group:]
+                    res = {}
+                    for i in range(group):
+                        res[f'group{i+1}'] = group_colors
+                    return res
+                
+            if type == 'axis':
+                seed = seed_axis % 5
+                if seed == 0:
+                    return {
+                        'axis': [extend_colors1[0]]
+                    }
+                if seed == 1:
+                    return {
+                        'axis': [extend_colors2[0]]
+                    }
+                if seed == 3: # gray
+                    return {
+                        'axis': [gray_color]
+                    }
+                if seed == 4: # main color
+                    return {
+                        'axis': [main_color_hex]
+                    }
+                return {
+                    'axis': [other_color]
+                }
+        
+        if self.mode == 'dual-color':
+            selected_color = self.main_color
+            color1 = selected_color[0]
+            color2 = selected_color[1]
+            
+            if number < 6:
+                extend_colors1_l = extend_color_in_l(color1)
+                extend_colors2_l = extend_color_in_l(color2)
+                extend_colors1_c = extend_color_in_c(color1)
+                extend_colors2_c = extend_color_in_c(color2)
+            else:
+                extend_colors1_l = extend_color_in_l(color1, number)
+                extend_colors2_l = extend_color_in_l(color2, number)
+                extend_colors1_c = extend_color_in_c(color1, number)
+                extend_colors2_c = extend_color_in_c(color2, number)
+
+            color1_hex = rgb_to_hex(*color1)
+            color2_hex = rgb_to_hex(*color2)
+            main_color1_hex = color1_hex
+            main_color2_hex = color2_hex
+
+            middle_colors = [color for color in self.pool['other_colors']]
+            # print("middle_colors: ", middle_colors)
+            if len(middle_colors) == 0:
+                middle_colors = ["#ababab"]
+            middle_color_seed = seed_middle_color % len(middle_colors)
+            middle_color = middle_colors[middle_color_seed]
+
+            other_color = None
+            gray_color = None
+            if self.lightness == 'dark':
+                other_color = self.basic_colors_hex[1]
+                gray_color = self.basic_colors_hex[4]
+            else:
+                other_color = self.basic_colors_hex[0]
+                gray_color = self.basic_colors_hex[2]
+            if seed_context_color > 0 and len(self.pool['context_color']) > 0:
+                choice = seed_context_color % len(self.pool['context_color'])
+                gray_color = self.pool['context_color'][choice]
+
+            if type == 'text':
+                seed = seed_text % 14
+                if seed == 0:
+                    return {
+                        'title': [other_color],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                if seed == 1:
+                    return {
+                        'title': [other_color],
+                        'caption': [gray_color],
+                        'annotation': [gray_color],
+                    }
+                if seed == 2:
+                    return {
+                        'title': [main_color1_hex],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                if seed == 3:
+                    return {
+                        'title': [main_color2_hex],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                if seed == 4:
+                    return {
+                        'title': [main_color1_hex],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [other_color],
+                    }
+                if seed == 5:
+                    return {
+                        'title': [main_color2_hex],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [other_color],
+                    }
+                if seed == 6:
+                    return {
+                        'title': [other_color],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [other_color],
+                    }
+                if seed == 7:
+                    return {
+                        'title': [main_color1_hex],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 8:
+                    return {
+                        'title': [main_color2_hex],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 9:
+                    return {
+                        'title': [other_color],
+                        'caption': [main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 10:
+                    return {
+                        'title': [main_color1_hex],
+                        'caption': [other_color, main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 11:
+                    return {
+                        'title': [main_color2_hex],
+                        'caption': [other_color, main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 12:
+                    return {
+                        'title': [other_color],
+                        'caption': [other_color, main_color1_hex, main_color2_hex],
+                        'annotation': [gray_color],
+                    }
+                if seed == 13:
+                    return {
+                        'title': [main_color1_hex, main_color2_hex, middle_color],
+                        'caption': [other_color],
+                        'annotation': [gray_color],
+                    }
+
+            if type == 'marks':
+                if group == 1 and number == 2:
+                    return {
+                        'group1': [main_color1_hex, main_color2_hex],
+                    }
+                if group == 1:
+                    seed = seed_mark % 4
+                    if seed == 0:
+                        return {
+                            'group1': [other_color],
+                        }
+                    if seed == 1:
+                        return {
+                            'group1': [main_color1_hex],
+                        }
+                    if seed == 2:
+                        return {
+                            'group1': [main_color2_hex],
+                        }
+                    if seed == 3:
+                        return {
+                            'group1': [gray_color],
+                        }
+                else:
+                    assert group == 2
+                    seed = seed_mark % 5
+                    if seed == 0:
+                        return {
+                            'group1': [main_color1_hex],
+                            'group2': [main_color2_hex],
+                        }
+                    if seed == 1:
+                        return {
+                            'group1': extend_colors1_l[:number],
+                            'group2': extend_colors2_l[:number],
+                        }
+                    if seed == 2:
+                        return {
+                            'group1': extend_colors1_l[-number:],
+                            'group2': extend_colors2_l[-number:],
+                        }
+                    if seed == 3:
+                        return {
+                            'group1': extend_colors1_c[:number],
+                            'group2': extend_colors2_c[:number],
+                        }
+                    if seed == 4:
+                        return {
+                            'group1': extend_colors1_c[-number:],
+                            'group2': extend_colors2_c[-number:],
+                        }
+
+            if type == 'axis':
+                axis_seed = seed_axis % 5
+                if axis_seed == 0:
+                    return {
+                        'axis': [other_color],
+                    }
+                if axis_seed == 1:
+                    return {
+                        'axis': [gray_color],
+                    }
+                if axis_seed == 2:
+                    return {
+                        'axis': [extend_colors1_l[0]],
+                    }
+                if axis_seed == 3:
+                    return {
+                        'axis': [extend_colors2_l[0]],
+                    }
+                if axis_seed == 4:
+                    return {
+                        'axis': [middle_color],
+                    }
+
+        if self.mode == 'colorful':
+            other_color = None
+            gray_color = None
+            if self.lightness == 'dark':
+                other_color = self.basic_colors_hex[1]
+                gray_color = self.basic_colors_hex[4]
+            else:
+                other_color = self.basic_colors_hex[0]
+                gray_color = self.basic_colors_hex[2]
+            if seed_context_color > 0 and len(self.pool['context_color']) > 0:
+                choice = seed_context_color % len(self.pool['context_color'])
+                gray_color = self.pool['context_color'][choice]
+            
+            if type == 'text':
+                seed = seed_text % 2
+                if seed == 0:
+                    return {
+                        'title': [other_color],
+                        'caption': [other_color],
+                        'annotation': [other_color],
+                    }
+                if self.middle_color is None:
+                    rand_color = random.choice(self.rgb_pool_hex)
+                    extend_colors = extend_color_in_l(hex_to_rgb(rand_color))
+                    dark_rand_color = extend_colors[0] if self.lightness == 'dark' else extend_colors[-1]
+                    self.middle_color = {
+                        'color': rand_color,
+                        'dark': dark_rand_color,
+                    }
+                return {
+                    'title': [self.middle_color['dark']],
+                    'caption': [self.middle_color['dark']],
+                    'annotation': [self.middle_color['dark']]
+                }
+            if type == 'marks':
+                seed = seed_mark % 2 + 1
+                if group == 1:
+                    clist = []
+                    if len(self.rgb_pool_hex) >= number:
+                        if seed == 1:
+                            clist = self.rgb_pool_hex[:number]
+                        if seed == 2:
+                            clist = self.rgb_pool_hex[-number:]
+                    elif len(self.rgb_pool_hex_2) >= number:
+                        if seed == 1:
+                            clist = self.rgb_pool_hex_2[:number]
+                        if seed == 2:
+                            clist = self.rgb_pool_hex_2[-number:]
+                    elif len(self.rgb_pool_hex_3) >= number:
+                        if seed == 1:
+                            clist = self.rgb_pool_hex_3[:number]
+                        if seed == 2:
+                            clist = self.rgb_pool_hex_3[-number:]
+                    else:
+                        clist = static_palettes.get_colors(number)
+                    return {
+                        'group1': clist,
+                    }
+                else:
+                    base_colors = []
+                    if len(self.rgb_pool_hex) >= group:
+                        if seed == 1:
+                            base_colors = self.rgb_pool_hex[:group]
+                        if seed == 2:
+                            base_colors = self.rgb_pool_hex[-group:]
+                    elif len(self.rgb_pool_hex_2) >= group:
+                        if seed == 1:
+                            base_colors = self.rgb_pool_hex_2[:group]
+                        if seed == 2:
+                            base_colors = self.rgb_pool_hex_2[-group:]
+                    elif len(self.rgb_pool_hex_3) >= group:
+                        if seed == 1:
+                            base_colors = self.rgb_pool_hex_3[:group]
+                        if seed == 2:
+                            base_colors = self.rgb_pool_hex_3[-group:]
+                    else:
+                        base_colors = static_palettes.get_colors(group)
+                    if seed_color % 2 == 0:
+                        res = {}
+                        for i in range(group):
+                            res[f'group{i+1}'] = [base_colors[i] for _ in range(number)]
+                    else:
+                        res = {}
+                        for i in range(group):
+                            res[f'group{i+1}'] = extend_color_in_l(hex_to_rgb(base_colors[i]), number)
+                    return res
+
+            if type == 'axis':
+                seed = seed_axis % 3
+                if seed == 0:
+                    return {
+                        'axis': [gray_color],
+                    }
+                if seed == 1:
+                    return {
+                        'axis': [other_color],
+                    }
+                return {
+                    'axis': [random.choice(self.rgb_pool_hex)]
+                }
+
+    def rank_color(self, palette, importance):
+        assert len(palette) == len(importance)
+        dists_to_bcg = [ciede2000(hex_to_rgb(color), self.bcg_color) for color in palette]
+        # sort by importance, high importance with high dist to bcg
+        # low importance with low dist to bcg
+        dist_sorted_indices = sorted(range(len(dists_to_bcg)), key=lambda i: dists_to_bcg[i], reverse=True)
+        imp_sorted_indices = sorted(range(len(importance)), key=lambda i: importance[i], reverse=True)
+
+        ranked_palette = [None] * len(palette)
+
+        for rank, color_idx in enumerate(dist_sorted_indices):
+            ranked_palette[imp_sorted_indices[rank]] = palette[color_idx]
+
+        return ranked_palette
+
+    def rank_color_by_contrast(self, palette):
+        dists_to_bcg = [ciede2000(hex_to_rgb(color), self.bcg_color) for color in palette]
+        print("bcg_color: ", self.bcg_color)
+        sorted_colors = sorted(zip(palette, dists_to_bcg), key=lambda x: -x[1])
+        ranked_palette = [color for color, _ in sorted_colors]
+        return ranked_palette

modules/color_recommender/create_index.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+import argparse
+import os
+from modules.color_recommender.color_index_builder import ColorIndexBuilder
+
+def main(force: bool=False):
+    parser = argparse.ArgumentParser(description='Build color palette index using FAISS and SentenceTransformer')
+    parser.add_argument('--input', '-i', type=str, default='./static/color_palette.json',
+                        help='Path to input color palette JSON file')
+    parser.add_argument('--output', '-o', type=str, default='./static/color_palette.index',
+                        help='Path to save the FAISS index')
+    parser.add_argument('--force', '-f', action='store_true',
+                        help='Force rebuild even if index exists')
+    parser.add_argument('--embed_model_path', type=str, default='', help='Path to sentence embedding model (optional)')
+
+    
+    args = parser.parse_args()
+    
+    # Check if input file exists
+    if not os.path.exists(args.input):
+        print(f"Error: Input file {args.input} does not exist")
+        return 1
+    
+    # Check if output file exists and handle force flag
+    if os.path.exists(args.output) and not (force or args.force):
+        print(f"Index file {args.output} already exists. Use --force to rebuild.")
+        return 0
+    
+    try:
+        # Initialize and build index
+        print("Initializing ColorIndexBuilder...")
+        index_builder = ColorIndexBuilder(args.input, args.embed_model_path)
+        
+        print("Building index...")
+        index_builder.build_index()
+        
+        print(f"Saving index to {args.output}...")
+        index_builder.save_index(args.output)
+        
+        print("Index built successfully!")
+        return 0
+        
+    except Exception as e:
+        print(f"Error building index: {str(e)}")
+        return 1
+
+if __name__ == '__main__':
+    exit(main()) 

modules/color_recommender/infographic_retrieve.py

@@ -0,0 +1,88 @@
+import json, os
+import torch
+import numpy as np
+from sentence_transformers import SentenceTransformer, util
+from config import sentence_transformer_path, infographic_library_path, infographic_image_path
+model_path = sentence_transformer_path
+library_path = infographic_library_path
+image_root_path = infographic_image_path
+
+class InfographicRetriever:
+    def __init__(self, model_path, library_path, image_path):
+        """
+        Initialize the image retriever with a knowledge base and embedding model
+
+        Args:
+            library_path (str): Path to the JSON knowledge base file
+            model_path (str): Path to the sentence transformer model
+        """
+        self.model_path = model_path
+        self.library_path = library_path
+        self.image_path = image_path
+
+        # Load knowledge base
+        with open(library_path, 'r', encoding='utf-8') as f:
+            self.knowledge_base = json.load(f)
+
+        # Load embedding model
+        self.embedding_model = SentenceTransformer(model_path)
+
+        # Pre-compute embeddings
+        self._prepare_embeddings()
+
+    def _combine_text_fields(self, item):
+        """Combine different text fields into a single string"""
+        return f"{item['title']}  {item['description']} {item['main_insight']}" + " ".join(item['columns'])
+
+    def _prepare_embeddings(self):
+        """Pre-compute embeddings for all items in knowledge base"""
+        self.knowledge_texts = []
+        self.knowledge_ids = []
+
+        for key, record in self.knowledge_base.items():
+            combined_text = self._combine_text_fields(record)
+            self.knowledge_texts.append(combined_text)
+            self.knowledge_ids.append(key)
+
+        with torch.no_grad():
+            self.knowledge_embeddings = self.embedding_model.encode(
+                self.knowledge_texts,
+                convert_to_tensor=True,
+                normalize_embeddings=True
+            )
+
+    def retrieve_similar_entries(self, query_text, top_k=3):
+        """
+        Retrieve similar images based on text query
+
+        Args:
+            query_text (str): Text query to search for
+            top_k (int): Number of results to return
+
+        Returns:
+            list: List of dictionaries containing similar entries
+        """
+        with torch.no_grad():
+            query_emb = self.embedding_model.encode(
+                query_text,
+                convert_to_tensor=True,
+                normalize_embeddings=True
+            )
+
+        cosine_scores = util.cos_sim(query_emb, self.knowledge_embeddings)[0]
+        top_results = torch.topk(cosine_scores, k=top_k)
+
+        results = []
+        for score_idx, score_val in zip(top_results.indices, top_results.values):
+            idx = score_idx.item()
+            similarity = score_val.item()
+            doc_id = self.knowledge_ids[idx]
+            results.append((os.path.join(self.image_path, doc_id + '.jpeg'), similarity, doc_id))
+        return results
+
+if __name__ == "__main__":
+    retriever = InfographicRetriever(model_path, library_path, image_root_path)
+
+    user_query = "Cat in a hat"
+    similar_entries = retriever.retrieve_similar_entries(user_query, top_k=10)
+    from IPython import embed; embed()

modules/color_recommender/llm_api.py

@@ -0,0 +1,154 @@
+from openai import OpenAI
+from PIL import Image
+import base64
+from io import BytesIO
+from config import client_key, base_url
+
+image_max_size = 512
+model_name = 'gpt-4o-mini'
+
+client = OpenAI(
+    api_key=client_key,
+    base_url=base_url
+)
+
+def resize_image(img, max_size=512):
+    width, height = img.size
+    ratio = min(max_size / width, max_size / height)
+    if ratio >= 1:
+        return img
+    new_width = int(width * ratio)
+    new_height = int(height * ratio)
+    resized_img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
+    return resized_img
+    
+def image_to_base64(image_path, target_size=image_max_size):
+    with Image.open(image_path) as img:
+        # img = img.resize(size)
+        img = resize_image(img, target_size)
+        buffered = BytesIO()
+        img.save(buffered, format="PNG")
+        img_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8')
+        return img_base64
+
+wwxxhh = 0
+def ask(prompt):
+    global wwxxhh
+    number_of_trials = 0
+    while number_of_trials < 5:
+        try:
+            response = client.chat.completions.create(
+              model=model_name,
+              messages=[
+                {
+                  "role": "user",
+                  "content": [
+                    {   
+                        "type": "text", 
+                        "text": prompt},
+                  ],
+                }
+              ]
+            )
+            wwxxhh += response.usage.total_tokens
+            return response.choices[0].message.content
+
+        except Exception as e:
+            number_of_trials += 1
+            print(e)
+
+    return 'Error!'
+
+def ask_image(prompt, image_data):
+    global wwxxhh
+    number_of_trials = 0
+    while number_of_trials < 5:
+        try:
+            response = client.chat.completions.create(
+              model=model_name,
+              messages=[
+                {
+                  "role": "user",
+                  "content": [
+                    {   
+                        "type": "text", 
+                        "text": prompt},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": f"data:image/jpeg;base64,{image_data}"
+                        },
+                    },
+                  ],
+                }
+              ]
+            )
+            wwxxhh += response.usage.total_tokens
+            return response.choices[0].message.content
+
+        except Exception as e:
+            number_of_trials += 1
+            print(e)
+
+    return 'Error!'
+
+def chat_with_image(prompts, image_data):
+    global wwxxhh
+    messages = []
+    
+    number_of_trials = 0
+    while number_of_trials < 5:
+        try:
+            messages.append({
+                "role": "user",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": prompts[0]
+                    },
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": f"data:image/jpeg;base64,{image_data}"
+                        }
+                    }
+                ]
+            })
+            
+            response = client.chat.completions.create(
+                model=model_name,
+                messages=messages
+            )
+            
+            messages.append({
+                "role": "assistant",
+                "content": response.choices[0].message.content
+            })
+            
+            wwxxhh += response.usage.total_tokens
+            
+            for prompt in prompts[1:]:
+                messages.append({
+                    "role": "user",
+                    "content": [{"type": "text", "text": prompt}]
+                })
+                
+                response = client.chat.completions.create(
+                    model=model_name,
+                    messages=messages
+                )
+                
+                messages.append({
+                    "role": "assistant",
+                    "content": response.choices[0].message.content
+                })
+                
+                wwxxhh += response.usage.total_tokens
+            
+            return [msg["content"] for msg in messages if msg["role"] == "assistant"]
+
+        except Exception as e:
+            number_of_trials += 1
+            print(e)
+            
+    return ['Error!'] * len(prompts)