import math import random import numpy as np from PIL import Image, ImageDraw, ImageFont from matplotlib.path import Path as MplPath from matplotlib.textpath import TextPath from matplotlib.transforms import Affine2D from . import config from .ewc import EfficientWordCloud from .fonts import get_cached_font, get_font_properties def normalize_relative_scores(values): if not values: return [] v_min = min(values) v_max = max(values) if math.isclose(v_min, v_max): return [1.0 for _ in values] scale = v_max - v_min return [(value - v_min) / scale for value in values] def build_log_rank_scores(freq_list, *, per_word=False): if not freq_list: return [] if per_word: word_weights = {} for word, freq in freq_list: f = max(float(freq), 1e-6) if word not in word_weights or f > word_weights[word]: word_weights[word] = f unique_weights = sorted(set(word_weights.values()), reverse=True) if len(unique_weights) <= 1: word_scores = {w: 1.0 for w in word_weights} else: log_vals = [math.log1p(w) for w in unique_weights] normed = normalize_relative_scores(log_vals) weight_to_score = dict(zip(unique_weights, normed)) word_scores = {w: weight_to_score[weight] for w, weight in word_weights.items()} return [word_scores.get(w, 1.0) for w, _ in freq_list] safe_freqs = [max(float(freq), 1e-6) for _word, freq in freq_list] log_scores = normalize_relative_scores([math.log1p(freq) for freq in safe_freqs]) rank_scores = [1.0 - (idx / max(1, len(freq_list) - 1)) for idx in range(len(freq_list))] total_ratio = config.LOG_WEIGHT_RATIO + config.RANK_WEIGHT_RATIO if total_ratio <= 0: return log_scores log_ratio = config.LOG_WEIGHT_RATIO / total_ratio rank_ratio = config.RANK_WEIGHT_RATIO / total_ratio return [ max(0.0, min(1.0, log_score * log_ratio + rank_score * rank_ratio)) for log_score, rank_score in zip(log_scores, rank_scores) ] def pick_palette_color(relative_score): if config.FONT_COLOR: return config.FONT_COLOR palette = config.LIGHT_COLOR_PALETTE if config.FILL_ON == "WHITE" else config.DARK_COLOR_PALETTE if not palette: return "#111111" idx = min(len(palette) - 1, max(0, int(round((1.0 - relative_score) * (len(palette) - 1))))) return palette[idx] def _build_layout_sequence(sorted_freq, max_words, layout_order_mode, layout_seed): if max_words <= 0 or not sorted_freq: return [] expanded_freq = list(sorted_freq) if len(expanded_freq) < max_words: base_words = expanded_freq[:] if not base_words: return [] while len(expanded_freq) < max_words: for item in base_words: if len(expanded_freq) >= max_words: break expanded_freq.append(item) expanded_freq = expanded_freq[:max_words] if layout_order_mode == config.LAYOUT_ORDER_MODE_SORTED or len(expanded_freq) <= 1: return expanded_freq band_count = min(3, len(expanded_freq)) band_size = math.ceil(len(expanded_freq) / band_count) bands = [] rng = random.Random(layout_seed) for band_idx in range(band_count): start = band_idx * band_size end = min(len(expanded_freq), start + band_size) band = expanded_freq[start:end] rng.shuffle(band) if band: bands.append(band) interleave_pattern = [0, 1, 0, 2] band_positions = [0] * len(bands) sequence = [] while len(sequence) < len(expanded_freq): appended = False for pattern_idx in interleave_pattern: if pattern_idx >= len(bands): continue pos = band_positions[pattern_idx] if pos >= len(bands[pattern_idx]): continue sequence.append(bands[pattern_idx][pos]) band_positions[pattern_idx] += 1 appended = True if len(sequence) >= len(expanded_freq): break if appended: continue for band_idx, band in enumerate(bands): pos = band_positions[band_idx] if pos < len(band): sequence.append(band[pos]) band_positions[band_idx] += 1 appended = True if len(sequence) >= len(expanded_freq): break if not appended: break return sequence class OptimizedEfficientWordCloud(EfficientWordCloud): def __init__(self, *args, large_font_ratio=config.LARGE_FONT_LIMIT_RATIO, size_scale=1.0, **kwargs): super().__init__(*args, **kwargs) self.large_font_ratio = large_font_ratio self.size_scale = size_scale def generate_from_frequencies(self, frequencies): if isinstance(frequencies, dict): freq_list = list(frequencies.items()) elif isinstance(frequencies, list): freq_list = frequencies else: raise ValueError("frequencies 必须是字典或 (word, freq) 列表") sorted_freq = sorted(freq_list, key=lambda x: x[1], reverse=True) layout_sequence = _build_layout_sequence( sorted_freq, self.max_words, config.LAYOUT_ORDER_MODE, config.LAYOUT_SEED, ) if not layout_sequence: return self self.layout_ = [] per_word_scores = build_log_rank_scores(freq_list, per_word=True) word_to_score = {} for (w, _f), s in zip(freq_list, per_word_scores): if w not in word_to_score or s > word_to_score[w]: word_to_score[w] = s score_by_index = [word_to_score.get(w, 1.0) for w, _ in layout_sequence] effective_max_font = max(self.min_font_size + 1, int(self.max_font_size * self.size_scale)) large_threshold = int(effective_max_font * config.LARGE_FONT_THRESHOLD_RATIO) if config.LIMIT_LARGE_FONTS else None large_limit = int(self.max_words * self.large_font_ratio) if config.LIMIT_LARGE_FONTS else None large_count = 0 rotation_flags = [np.random.random() > self.prefer_horizontal for _ in layout_sequence] # Dummy draw for textbbox measurement (no actual PIL image needed during placement) _measure_img = Image.new("L", (1, 1)) _measure_draw = ImageDraw.Draw(_measure_img) base_span = max(1, self.max_font_size - self.min_font_size) target_font_sizes = [] for score in score_by_index: raw_size = self.min_font_size + base_span * score f_size = max(config.MIN_FONT_FLOOR, int(round(raw_size * self.size_scale))) target_font_sizes.append(f_size) gap_fill_list = [] # 收集未成功放置的词,用于第二轮填充 for idx, (word, _freq) in enumerate(layout_sequence): font_size = target_font_sizes[idx] if config.LIMIT_LARGE_FONTS and large_threshold is not None and large_limit is not None: if font_size >= large_threshold and large_count >= large_limit: font_size = max(self.min_font_size, int(large_threshold * config.LARGE_FONT_CAP_RATIO)) current_size = font_size min_attempt_size = max(self.min_font_size, int(current_size * 0.4)) placed = False while current_size >= min_attempt_size: orientation = None rotate = rotation_flags[idx] if rotate: orientation = Image.ROTATE_90 font = get_cached_font(self.font_path, current_size) if orientation: transposed = ImageFont.TransposedFont(font, orientation=orientation) else: transposed = font bbox = _measure_draw.textbbox((0, 0), word, font=transposed) w_text = bbox[2] - bbox[0] h_text = bbox[3] - bbox[1] query_w = w_text + self.margin query_h = h_text + self.margin pos = self.grid.query_direct(query_h, query_w, np.random.randint(0, 2**31)) if pos is not None: y, x = pos draw_y = y + self.margin // 2 draw_x = x + self.margin // 2 # Stamp glyph bitmap into C++ canvas for pixel-accurate collision font = get_cached_font(self.font_path, current_size) if orientation: transposed = ImageFont.TransposedFont(font, orientation=orientation) else: transposed = font glyph_mask = transposed.getmask(word, mode="L") gw, gh = glyph_mask.size glyph_arr = np.frombuffer(bytes(glyph_mask), dtype=np.uint8).reshape(gh, gw) self.grid.stamp_and_rebuild(glyph_arr, gh, gw, draw_y, draw_x) color = pick_palette_color(score_by_index[idx]) self.layout_.append((word, current_size, (draw_y, draw_x), orientation, color)) if config.LIMIT_LARGE_FONTS and large_threshold is not None and current_size >= large_threshold: large_count += 1 placed = True break current_size -= 2 if not placed: gap_fill_list.append((word, score_by_index[idx])) # ── Gap-filling pass: 用更小的字号填充剩余空隙 ────────────── if gap_fill_list: gap_font_size = max(config.MIN_FONT_FLOOR, int(self.min_font_size * 0.8)) if gap_font_size >= config.MIN_FONT_FLOOR: placed_gap = 0 for word, score in gap_fill_list: font = get_cached_font(self.font_path, gap_font_size) bbox = _measure_draw.textbbox((0, 0), word, font=font) w_text = bbox[2] - bbox[0] h_text = bbox[3] - bbox[1] query_w = w_text + self.margin query_h = h_text + self.margin pos = self.grid.query_direct(query_h, query_w, np.random.randint(0, 2**31)) if pos is not None: y, x = pos draw_y = y + self.margin // 2 draw_x = x + self.margin // 2 glyph_mask = font.getmask(word, mode="L") gw, gh = glyph_mask.size glyph_arr = np.frombuffer(bytes(glyph_mask), dtype=np.uint8).reshape(gh, gw) self.grid.stamp_and_rebuild(glyph_arr, gh, gw, draw_y, draw_x) color = pick_palette_color(score) self.layout_.append((word, gap_font_size, (draw_y, draw_x), None, color)) placed_gap += 1 if placed_gap > 0: config._warn(f"Gap-filling: 用小字号 {gap_font_size} 额外放置了 {placed_gap}/{len(gap_fill_list)} 个词") return self def to_image(self): img = Image.new(self.mode, (self.width, self.height), self.background_color) draw = ImageDraw.Draw(img) for word, size, (y, x), orient, color in self.layout_: font = get_cached_font(self.font_path, size) if orient: font = ImageFont.TransposedFont(font, orientation=orient) draw.text((x, y), word, font=font, fill=color) return img def to_svg(self, filename): background = self.background_color with open(filename, "w", encoding="utf-8") as f: f.write( f'\n' ) f.write(f'\n') for word, size, (y, x), orient, color in self.layout_: try: path, tx, ty, _ = build_svg_text_path(word, size, x, y, self.font_path, orient) except Exception as exc: config._warn(f"SVG path 导出失败,跳过词条: {word}, error={exc}") continue f.write(f'\n') f.write("\n") def to_svg_stroke(self, filename, stroke_color="#000000", stroke_width=1.0): """生成描边版 SVG,适合激光雕刻机使用(描边路径,无填充)。""" with open(filename, "w", encoding="utf-8") as f: f.write( f'\n' ) f.write(f'\n') for word, size, (y, x), orient, _color in self.layout_: try: path, tx, ty, _ = build_svg_text_path(word, size, x, y, self.font_path, orient) except Exception as exc: config._warn(f"SVG stroke path 导出失败,跳过词条: {word}, error={exc}") continue f.write( f'\n' ) f.write("\n") def to_svg_dotfill(self, filename, dot_spacing=10, dot_radius=2, dot_color="#000000"): """生成点阵填充 SVG:文字区域用密排小圆点填充,适合激光雕刻逐点打标。""" from .render import render_layout_occupancy occ = render_layout_occupancy(self.layout_, (self.height, self.width), self.font_path) occ_arr = np.array(occ) with open(filename, "w", encoding="utf-8") as f: f.write( f'\n' ) f.write(f'\n') half = dot_spacing / 2 dot_count = 0 h, w = occ_arr.shape for gy in range(0, h, dot_spacing): for gx in range(0, w, dot_spacing): cy = min(gy + int(half), h - 1) cx = min(gx + int(half), w - 1) if occ_arr[cy, cx]: f.write( f'\n' ) dot_count += 1 f.write("\n") return dot_count def to_svg_custom(self, filename, fill_mode="fill", do_stroke=False, dot_spacing=10, dot_radius=2, color="#000000", line_spacing=6, line_width=1, line_angle=0, ring_radius=3, ring_width=1, ring_spacing=8): """统一 SVG 导出:fill_mode=fill|dot|line|ring,可叠加描边。""" # 预先构建所有文字路径(fill / dot 模式共用) text_paths = [] for word, size, (y, x), orient, _color in self.layout_: try: path, tx, ty, _ = build_svg_text_path(word, size, x, y, self.font_path, orient) text_paths.append((path, tx, ty)) except Exception as exc: config._warn(f"SVG path 导出失败,跳过: {word}, error={exc}") with open(filename, "w", encoding="utf-8") as f: f.write( f'\n' ) f.write(f'\n') if fill_mode == "dot": # 点阵模式:用 SVG pattern 平铺圆点 + clipPath 裁剪到文字形状 f.write('\n') f.write(f' \n') half = dot_spacing / 2 f.write(f' \n') f.write(' \n') self._write_text_clip(f, text_paths) f.write('\n') f.write(f'\n') elif fill_mode == "line": # 线条填充:用 matplotlib Path 渲染占用蒙版(与 SVG 完全对齐) import math as _m occ = render_path_occupancy(self.layout_, (self.height, self.width), self.font_path) angle = line_angle % 360 rad = _m.radians(angle) cos_a, sin_a = _m.cos(rad), _m.sin(rad) h, w = occ.shape step = 1 # 逐像素采样,保证线段连续 # 垂直方向的总范围(确保覆盖整个画布) perp_max = abs(h * cos_a) + abs(w * sin_a) n_lines = max(1, int(perp_max / line_spacing) + 1) sw = f'{line_width:g}' path_parts = [] for i in range(n_lines): d0 = (i - n_lines // 2) * line_spacing sx = -d0 * sin_a sy = d0 * cos_a n_steps = int(perp_max) + 1 run_start = None for s in range(n_steps + 1): px = sx + s * step * cos_a py = sy + s * step * sin_a ix, iy = int(round(px)), int(round(py)) inside = (0 <= iy < h and 0 <= ix < w and occ[iy, ix]) if inside: if run_start is None: run_start = (px, py) else: if run_start is not None: ex = px - step * cos_a ey = py - step * sin_a path_parts.append(f'M{run_start[0]:.1f} {run_start[1]:.1f}L{ex:.1f} {ey:.1f}') run_start = None if run_start is not None: ex = sx + n_steps * step * cos_a ey = sy + n_steps * step * sin_a path_parts.append(f'M{run_start[0]:.1f} {run_start[1]:.1f}L{ex:.1f} {ey:.1f}') if path_parts: f.write(f'\n') elif fill_mode == "ring": # 空心圆点填充:闭合路径,激光机可描一圈 occ = render_path_occupancy(self.layout_, (self.height, self.width), self.font_path) h, w = occ.shape r = ring_radius sw = f'{ring_width:g}' circle_parts = [] for gy in range(r, h - r, ring_spacing): for gx in range(r, w - r, ring_spacing): if not occ[gy, gx]: continue lx = gx - r rx = gx + r circle_parts.append( f'M{lx} {gy}A{r} {r} 0 1 0 {rx} {gy}A{r} {r} 0 1 0 {lx} {gy}Z' ) if circle_parts: f.write(f'\n') # 只有 fill 模式和显式描边时才输出 matplotlib 文字路径 # ring/line 模式用 PIL occupancy mask 生成填充,不需要文字轮廓 if fill_mode == "fill" or do_stroke: for path, tx, ty in text_paths: fill_attr = color if fill_mode == "fill" else "none" stroke_attr = f'stroke="{color}" stroke-width="1" stroke-linejoin="round" stroke-linecap="round"' if do_stroke else "" f.write(f'\n') f.write("\n") @staticmethod def _write_text_clip(f, text_paths): """将文字路径写入 (调用方负责 开闭)。""" f.write(' \n') for path, tx, ty in text_paths: f.write(f' \n') f.write(' \n') def build_svg_text_path(word, size, x, y, font_path, orient): path = TextPath((0, 0), word, prop=get_font_properties(font_path, size), size=size) if orient: path = path.transformed(Affine2D().rotate_deg(-90)) bbox = path.get_extents() tx = x - bbox.xmin ty = y + bbox.ymax # 返回变换后的 Path(已定位到画布坐标)以及 SVG 用的偏移量 transformed = path.transformed(Affine2D().scale(1, -1).translate(tx, ty)) return mpl_path_to_svg_d(path), tx, ty, transformed def mpl_path_to_svg_d(path): parts = [] for vertices, code in path.iter_segments(): if code == MplPath.MOVETO: x, y = vertices parts.append(f"M{x:.3f} {y:.3f}") elif code == MplPath.LINETO: x, y = vertices parts.append(f"L{x:.3f} {y:.3f}") elif code == MplPath.CURVE3: x1, y1, x2, y2 = vertices parts.append(f"Q{x1:.3f} {y1:.3f} {x2:.3f} {y2:.3f}") elif code == MplPath.CURVE4: x1, y1, x2, y2, x3, y3 = vertices parts.append( f"C{x1:.3f} {y1:.3f} {x2:.3f} {y2:.3f} {x3:.3f} {y3:.3f}" ) elif code == MplPath.CLOSEPOLY: parts.append("Z") return " ".join(parts) def render_path_occupancy(layout_data, canvas_shape, font_path): """渲染文字占用蒙版:字形笔画=1,字内空洞(如口)=0,外部=0。 使用 PIL 渲染文字蒙版(与画布坐标完全对齐)+ 边界泛洪填充来区分外部区域与字内空洞。 layout_data: [(word, size, (y, x), orient, color), ...] 同 self.layout_ """ from collections import deque h, w = canvas_shape if not layout_data: return np.zeros((h, w), dtype=np.uint8) # 用 PIL 渲染文字蒙版(坐标系与 to_image() 完全一致) mask = Image.new("L", (w, h), 0) draw = ImageDraw.Draw(mask) for word, size, (y, x), orient, _color in layout_data: font = get_cached_font(font_path, size) if orient: font = ImageFont.TransposedFont(font, orientation=orient) draw.text((x, y), word, font=font, fill=255) occ_raw = (np.array(mask) > 127).astype(np.uint8) # 泛洪填充:从边框出发标记所有与外部连通的白色区域 # 口 等闭合字符的内部空洞不会与边框连通,因此正确保留为空 outside = np.zeros_like(occ_raw, dtype=np.uint8) q = deque() for x in range(w): if occ_raw[0, x]: q.append((0, x)) outside[0, x] = 1 if occ_raw[h - 1, x]: q.append((h - 1, x)) outside[h - 1, x] = 1 for y in range(1, h - 1): if occ_raw[y, 0]: q.append((y, 0)) outside[y, 0] = 1 if occ_raw[y, w - 1]: q.append((y, w - 1)) outside[y, w - 1] = 1 while q: cy, cx = q.popleft() for dy, dx in ((-1, 0), (1, 0), (0, -1), (0, 1)): ny, nx = cy + dy, cx + dx if 0 <= ny < h and 0 <= nx < w and occ_raw[ny, nx] and not outside[ny, nx]: outside[ny, nx] = 1 q.append((ny, nx)) # 最终蒙版:文字笔画=1,外部和字内空洞=0 return (occ_raw & (~outside).astype(np.uint8)).astype(np.uint8)