from __future__ import annotations import argparse import gc import hashlib import json import math import random from collections import deque from heapq import heappop, heappush from pathlib import Path from PIL import Image, ImageChops, ImageDraw, ImageEnhance, ImageFilter, ImageOps Image.MAX_IMAGE_PIXELS = None ROOT = Path(__file__).resolve().parents[1] DEFAULT_DATA = ROOT / "tmp" / "battle-map-v2-runtime-data.json" SOURCE_DIR = ROOT / "src" / "assets" / "images" / "battle" / "terrain-v2" DOCS_DIR = ROOT / "docs" SOURCE_PATHS = { kind: SOURCE_DIR / f"{kind}-v2-source.png" for kind in ("plain", "road", "hill", "forest", "river", "cliff", "village", "fort", "camp") } REGION_KINDS = ("hill", "forest", "cliff") STRUCTURE_KINDS = {"village", "fort", "camp"} CONNECTABLE_ROAD = {"road", "village", "fort", "camp"} MAX_TEXTURE_EDGE = 8192 WATER_LAYER_REASSERT_BATTLES = {40, 41, 42, 43, 51, 52, 53, 54, 59, 60, 61, 62, 63} GRID_CARDINAL_NEIGHBORS = ((1, 0), (-1, 0), (0, 1), (0, -1)) GRID_EIGHT_NEIGHBORS = ((0, -1), (1, -1), (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1)) def coordinate_hash(x: int, y: int, seed: int, salt: int = 0) -> int: value = (x * 0x1F123BB5) ^ (y * 0x5F356495) ^ (seed * 0x9E3779B1) ^ salt value &= 0xFFFFFFFF value ^= value >> 16 value = (value * 0x7FEB352D) & 0xFFFFFFFF value ^= value >> 15 value = (value * 0x846CA68B) & 0xFFFFFFFF value ^= value >> 16 return value def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Generate topology-safe V2 battle maps from AI-authored terrain sources.") parser.add_argument("--data", type=Path, default=DEFAULT_DATA) parser.add_argument("--maps", default="2-66", help="Battle numbers, ranges, or comma-separated values (default: 2-66).") parser.add_argument("--quality", type=int, default=84) parser.add_argument("--contact-sheet", type=Path, default=DOCS_DIR / "battle-map-v2-contact-sheet.jpg") parser.add_argument("--manifest", type=Path, default=DOCS_DIR / "battle-map-v2-manifest.json") parser.add_argument("--skip-artifacts", action="store_true", help="Skip manifest/contact sheet generation during a targeted preview.") return parser.parse_args() def parse_number_set(value: str) -> set[int]: result: set[int] = set() for part in value.split(","): token = part.strip() if not token: continue if "-" in token: start_text, end_text = token.split("-", 1) start, end = int(start_text), int(end_text) result.update(range(min(start, end), max(start, end) + 1)) else: result.add(int(token)) return result def load_sources() -> dict[str, Image.Image]: missing = [str(path) for path in SOURCE_PATHS.values() if not path.is_file()] if missing: raise FileNotFoundError(f"Missing V2 terrain sources: {', '.join(missing)}") return {kind: Image.open(path).convert("RGB") for kind, path in SOURCE_PATHS.items()} def output_tile_size(columns: int, rows: int) -> int: longest = max(columns, rows) if longest <= 80: tile = 96 elif longest <= 90: tile = 88 elif longest <= 100: tile = 80 elif longest <= 112: tile = 72 elif longest <= 128: tile = 64 else: tile = 60 if longest * tile > MAX_TEXTURE_EDGE: tile = MAX_TEXTURE_EDGE // longest return tile def feathered_patch_mask(side: int, alpha: int = 236) -> Image.Image: fade = max(24, side // 7) mask = Image.new("L", (side, side), 0) draw = ImageDraw.Draw(mask) inset = max(4, fade // 3) draw.rounded_rectangle( (inset, inset, side - inset, side - inset), radius=max(16, fade), fill=alpha, ) return mask.filter(ImageFilter.GaussianBlur(max(8, fade // 2))) def fill_texture(source: Image.Image, size: tuple[int, int], tile: int, kind: str, seed: int) -> Image.Image: """Synthesize a full material without visible square tiling. A low-frequency color field supplies continuity while overlapping, feathered source crops restore game-scale detail. This avoids stamping the source's macro composition across large late-game maps. """ rng = random.Random(seed) width, height = size average_sample = source.resize((1, 1), Image.Resampling.BOX) average = average_sample.getpixel((0, 0)) average_sample.close() dark = tuple(max(0, channel - (31 if kind == "river" else 26)) for channel in average) light = tuple(min(255, channel + (23 if kind == "river" else 29)) for channel in average) noise_side = 768 noise_rng = random.Random(seed ^ 0x5F3759DF) noise = Image.frombytes("L", (noise_side, noise_side), noise_rng.randbytes(noise_side * noise_side)) noise = noise.filter(ImageFilter.GaussianBlur(2.2 if kind == "river" else 3.5)) noise = noise.resize(size, Image.Resampling.BICUBIC) result = ImageOps.colorize(noise, dark, light) noise.close() patch_side = min(1152, max(512, tile * (17 if kind == "river" else 14))) stride = max(192, round(patch_side * 0.58)) variants: list[Image.Image] = [] for index in range(18): crop_side = round(source.width * rng.uniform(0.46, 0.82)) left = rng.randint(0, max(0, source.width - crop_side)) top = rng.randint(0, max(0, source.height - crop_side)) crop = source.crop((left, top, left + crop_side, top + crop_side)) variant = ImageOps.fit(crop, (patch_side, patch_side), method=Image.Resampling.LANCZOS) crop.close() if index % 4 == 1: variant = ImageOps.mirror(variant) elif index % 4 == 2: variant = variant.transpose(Image.Transpose.ROTATE_180) elif index % 4 == 3 and kind != "river": variant = ImageOps.flip(variant) if kind == "river": variant = ImageEnhance.Brightness(variant).enhance(rng.uniform(0.95, 1.05)) variant = ImageEnhance.Color(variant).enhance(rng.uniform(0.95, 1.04)) else: variant = ImageEnhance.Brightness(variant).enhance(rng.uniform(0.985, 1.018)) variant = ImageEnhance.Color(variant).enhance(rng.uniform(0.98, 1.02)) variants.append(variant) patch_mask = feathered_patch_mask(patch_side, 228 if kind == "river" else 148) row = 0 for base_y in range(-patch_side // 2, height + patch_side // 2, stride): offset = stride // 2 if row % 2 else 0 for base_x in range(-patch_side // 2 - offset, width + patch_side // 2, stride): jitter = max(12, stride // 7) x = base_x + rng.randint(-jitter, jitter) y = base_y + rng.randint(-jitter, jitter) result.paste(variants[rng.randrange(len(variants))], (x, y), patch_mask) row += 1 patch_mask.close() for variant in variants: variant.close() return result def is_kind(terrain: list[list[str]], x: int, y: int, kind: str) -> bool: return 0 <= y < len(terrain) and 0 <= x < len(terrain[0]) and terrain[y][x] == kind def region_mask(terrain: list[list[str]], kind: str, tile: int, seed: int) -> Image.Image: width = len(terrain[0]) * tile height = len(terrain) * tile rng = random.Random(seed * 97 + sum(ord(char) for char in kind)) mask = Image.new("L", (width, height), 0) draw = ImageDraw.Draw(mask) for y, row in enumerate(terrain): for x, terrain_kind in enumerate(row): if terrain_kind != kind: continue jitter = tile // (14 if kind == "river" else (10 if kind == "cliff" else 7)) cx = x * tile + tile // 2 + rng.randint(-jitter, jitter) cy = y * tile + tile // 2 + rng.randint(-jitter, jitter) if kind == "river": radius_min, radius_max = (0.63, 0.74) elif kind == "cliff": radius_min, radius_max = (0.51, 0.63) else: radius_min, radius_max = (0.49, 0.67) rx = round(tile * rng.uniform(radius_min, radius_max)) ry = round(tile * rng.uniform(radius_min, radius_max)) draw.ellipse((cx - rx, cy - ry, cx + rx, cy + ry), fill=255) core = max(4, round(tile * 0.22)) center_x = x * tile + tile // 2 center_y = y * tile + tile // 2 draw.ellipse((center_x - core, center_y - core, center_x + core, center_y + core), fill=255) for dx, dy in ((1, 0), (0, 1)): if not is_kind(terrain, x + dx, y + dy, kind): continue ncx = (x + dx) * tile + tile // 2 ncy = (y + dy) * tile + tile // 2 if kind == "river": half_width = round(tile * rng.uniform(0.57, 0.68)) else: half_width = round(tile * rng.uniform(0.40, 0.57)) if dx: draw.polygon( [ (cx, cy - half_width), (ncx, ncy - half_width + rng.randint(-tile // 14, tile // 14)), (ncx, ncy + half_width + rng.randint(-tile // 14, tile // 14)), (cx, cy + half_width), ], fill=255, ) else: draw.polygon( [ (cx - half_width, cy), (ncx - half_width + rng.randint(-tile // 14, tile // 14), ncy), (ncx + half_width + rng.randint(-tile // 14, tile // 14), ncy), (cx + half_width, cy), ], fill=255, ) blur = max(1.0, tile * (0.035 if kind == "river" else 0.045)) return mask.filter(ImageFilter.GaussianBlur(blur)) def terrain_components(terrain: list[list[str]], kind: str) -> list[list[tuple[int, int]]]: cells = {(x, y) for y, row in enumerate(terrain) for x, value in enumerate(row) if value == kind} components: list[list[tuple[int, int]]] = [] while cells: start = cells.pop() component = [start] stack = [start] while stack: x, y = stack.pop() for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): neighbor = (x + dx, y + dy) if neighbor in cells: cells.remove(neighbor) component.append(neighbor) stack.append(neighbor) components.append(component) return components def component_grid_distances(component: list[tuple[int, int]]) -> dict[tuple[int, int], int]: """Return a cheap Manhattan distance-to-bank field on terrain cells.""" cells = set(component) distances: dict[tuple[int, int], int] = {} pending: deque[tuple[int, int]] = deque() for x, y in cells: if any((x + dx, y + dy) not in cells for dx, dy in GRID_CARDINAL_NEIGHBORS): distances[(x, y)] = 1 pending.append((x, y)) while pending: x, y = pending.popleft() next_distance = distances[(x, y)] + 1 for dx, dy in GRID_CARDINAL_NEIGHBORS: neighbor = (x + dx, y + dy) if neighbor in cells and neighbor not in distances: distances[neighbor] = next_distance pending.append(neighbor) return distances def thin_grid_component(component: list[tuple[int, int]]) -> set[tuple[int, int]]: """Topology-preserving Zhang-Suen thinning on the small terrain grid.""" skeleton = set(component) changed = True while changed: changed = False for phase in (0, 1): removed: list[tuple[int, int]] = [] for x, y in skeleton: neighbors = [(x + dx, y + dy) in skeleton for dx, dy in GRID_EIGHT_NEIGHBORS] neighbor_count = sum(neighbors) transitions = sum( not neighbors[index] and neighbors[(index + 1) % 8] for index in range(8) ) if not 2 <= neighbor_count <= 6 or transitions != 1: continue north, _, east, _, south, _, west, _ = neighbors if phase == 0: if (north and east and south) or (east and south and west): continue elif (north and east and west) or (north and south and west): continue removed.append((x, y)) if removed: skeleton.difference_update(removed) changed = True return skeleton def skeleton_adjacency( skeleton: set[tuple[int, int]], ) -> dict[tuple[int, int], tuple[tuple[int, int], ...]]: return { point: tuple( sorted( (point[0] + dx, point[1] + dy) for dx, dy in GRID_EIGHT_NEIGHBORS if (point[0] + dx, point[1] + dy) in skeleton ) ) for point in skeleton } def skeleton_junction_clusters( adjacency: dict[tuple[int, int], tuple[tuple[int, int], ...]], ) -> list[set[tuple[int, int]]]: remaining = {point for point, neighbors in adjacency.items() if len(neighbors) >= 3} clusters: list[set[tuple[int, int]]] = [] while remaining: start = remaining.pop() cluster = {start} pending = [start] while pending: x, y = pending.pop() for dx, dy in GRID_EIGHT_NEIGHBORS: neighbor = (x + dx, y + dy) if neighbor in remaining: remaining.remove(neighbor) cluster.add(neighbor) pending.append(neighbor) clusters.append(cluster) return clusters def extend_skeleton_endpoint( path: list[tuple[int, int]], component_cells: set[tuple[int, int]], from_start: bool, ) -> tuple[float, float]: endpoint = path[0] if from_start else path[-1] inner_index = min(3, len(path) - 1) inner = path[inner_index] if from_start else path[-inner_index - 1] dx = (endpoint[0] > inner[0]) - (endpoint[0] < inner[0]) dy = (endpoint[1] > inner[1]) - (endpoint[1] < inner[1]) if dx == 0 and dy == 0 and len(path) >= 2: adjacent = path[1] if from_start else path[-2] dx = (endpoint[0] > adjacent[0]) - (endpoint[0] < adjacent[0]) dy = (endpoint[1] > adjacent[1]) - (endpoint[1] < adjacent[1]) current = endpoint while dx or dy: candidate = (current[0] + dx, current[1] + dy) if candidate not in component_cells: break current = candidate return current[0] + 0.5 + dx * 0.5, current[1] + 0.5 + dy * 0.5 def trace_skeleton_chains( skeleton: set[tuple[int, int]], component: list[tuple[int, int]], ) -> tuple[ list[tuple[list[tuple[float, float]], bool, bool, bool]], list[tuple[float, float]], int, ]: """Contract junction pixels and return endpoint-to-junction skeleton chains.""" adjacency = skeleton_adjacency(skeleton) endpoint_cells = {point for point, neighbors in adjacency.items() if len(neighbors) <= 1} junction_clusters = skeleton_junction_clusters(adjacency) junction_index: dict[tuple[int, int], int] = {} junction_points: list[tuple[float, float]] = [] for index, cluster in enumerate(junction_clusters): for point in cluster: junction_index[point] = index junction_points.append( ( sum(x + 0.5 for x, _ in cluster) / len(cluster), sum(y + 0.5 for _, y in cluster) / len(cluster), ) ) critical = endpoint_cells | set(junction_index) visited_edges: set[tuple[tuple[int, int], tuple[int, int]]] = set() component_cells = set(component) chains: list[tuple[list[tuple[float, float]], bool, bool, bool]] = [] def edge_key( first: tuple[int, int], second: tuple[int, int] ) -> tuple[tuple[int, int], tuple[int, int]]: return (first, second) if first <= second else (second, first) for start in sorted(critical): for neighbor in adjacency[start]: first_edge = edge_key(start, neighbor) if first_edge in visited_edges: continue visited_edges.add(first_edge) path = [start, neighbor] previous, current = start, neighbor while current not in critical: candidates = [candidate for candidate in adjacency[current] if candidate != previous] if not candidates: break following = candidates[0] next_edge = edge_key(current, following) if next_edge in visited_edges: break visited_edges.add(next_edge) path.append(following) previous, current = current, following start_junction = junction_index.get(path[0]) end_junction = junction_index.get(path[-1]) if start_junction is not None and start_junction == end_junction and len(path) <= 2: continue points = [(x + 0.5, y + 0.5) for x, y in path] if start_junction is not None: points[0] = junction_points[start_junction] if end_junction is not None: points[-1] = junction_points[end_junction] start_is_endpoint = path[0] in endpoint_cells end_is_endpoint = path[-1] in endpoint_cells if start_is_endpoint: points[0] = extend_skeleton_endpoint(path, component_cells, True) if end_is_endpoint: points[-1] = extend_skeleton_endpoint(path, component_cells, False) if len(points) >= 2 and points[0] != points[-1]: chains.append((points, start_is_endpoint, end_is_endpoint, False)) # A ring has no critical cell, so trace any remaining degree-two cycle once. for start in sorted(skeleton): for neighbor in adjacency[start]: first_edge = edge_key(start, neighbor) if first_edge in visited_edges: continue visited_edges.add(first_edge) path = [start, neighbor] previous, current = start, neighbor while current != start: candidates = [ candidate for candidate in adjacency[current] if candidate != previous and edge_key(current, candidate) not in visited_edges ] if not candidates: break following = candidates[0] visited_edges.add(edge_key(current, following)) path.append(following) previous, current = current, following closed = path[-1] == start grid_points = [(x + 0.5, y + 0.5) for x, y in (path[:-1] if closed else path)] if len(grid_points) >= 2: chains.append((grid_points, False, False, closed)) return chains, junction_points, len(endpoint_cells) def polyline_tortuosity(points: list[tuple[float, float]]) -> float: if len(points) < 2: return math.inf length = polyline_length(points) chord = math.hypot(points[-1][0] - points[0][0], points[-1][1] - points[0][1]) return length / chord if chord > 1e-6 else math.inf def polyline_length(points: list[tuple[float, float]]) -> float: return sum( math.hypot(second[0] - first[0], second[1] - first[1]) for first, second in zip(points, points[1:]) ) def skeleton_node_key(point: tuple[float, float]) -> tuple[float, float]: return round(point[0], 6), round(point[1], 6) def skeleton_chain_graph( chains: list[tuple[list[tuple[float, float]], bool, bool, bool]], ) -> tuple[ dict[tuple[float, float], list[tuple[tuple[float, float], int, float]]], dict[tuple[float, float], tuple[float, float]], ]: adjacency: dict[ tuple[float, float], list[tuple[tuple[float, float], int, float]] ] = {} node_points: dict[tuple[float, float], tuple[float, float]] = {} for index, (points, _, _, closed) in enumerate(chains): if closed or len(points) < 2: continue start = skeleton_node_key(points[0]) end = skeleton_node_key(points[-1]) if start == end: continue node_points.setdefault(start, points[0]) node_points.setdefault(end, points[-1]) weight = max(1e-6, polyline_length(points)) adjacency.setdefault(start, []).append((end, index, weight)) adjacency.setdefault(end, []).append((start, index, weight)) for edges in adjacency.values(): edges.sort(key=lambda edge: (edge[0], edge[1])) return adjacency, node_points def weighted_skeleton_paths( adjacency: dict[ tuple[float, float], list[tuple[tuple[float, float], int, float]] ], start: tuple[float, float], ) -> tuple[ dict[tuple[float, float], float], dict[tuple[float, float], tuple[tuple[float, float], int]], ]: distances = {start: 0.0} parents: dict[tuple[float, float], tuple[tuple[float, float], int]] = {} pending = [(0.0, start)] while pending: distance, current = heappop(pending) if distance > distances[current] + 1e-9: continue for neighbor, edge_index, weight in adjacency.get(current, []): candidate = distance + weight if candidate + 1e-9 >= distances.get(neighbor, math.inf): continue distances[neighbor] = candidate parents[neighbor] = current, edge_index heappush(pending, (candidate, neighbor)) return distances, parents def skeleton_path_to( parents: dict[tuple[float, float], tuple[tuple[float, float], int]], start: tuple[float, float], end: tuple[float, float], ) -> tuple[list[tuple[float, float]], list[int]]: nodes = [end] edges: list[int] = [] current = end while current != start: parent = parents.get(current) if parent is None: return [], [] current, edge_index = parent nodes.append(current) edges.append(edge_index) nodes.reverse() edges.reverse() return nodes, edges def active_skeleton_chains( chains: list[tuple[list[tuple[float, float]], bool, bool, bool]], kept_indices: set[int] | None = None, ) -> tuple[ list[tuple[list[tuple[float, float]], bool, bool, bool]], list[tuple[float, float]], int, ]: kept = set(range(len(chains))) if kept_indices is None else kept_indices adjacency, node_points = skeleton_chain_graph( [chain for index, chain in enumerate(chains) if index in kept] ) degrees = {node: len(edges) for node, edges in adjacency.items()} active_chains: list[tuple[list[tuple[float, float]], bool, bool, bool]] = [] for index, (points, _, _, closed) in enumerate(chains): if index not in kept: continue if closed: active_chains.append((points, False, False, True)) continue start = skeleton_node_key(points[0]) end = skeleton_node_key(points[-1]) active_chains.append((points, degrees.get(start) == 1, degrees.get(end) == 1, False)) junction_points = sorted( (node_points[node] for node, degree in degrees.items() if degree >= 3), key=lambda point: (point[1], point[0]), ) endpoint_count = sum(degree == 1 for degree in degrees.values()) return active_chains, junction_points, endpoint_count def prune_skeleton_branches( chains: list[tuple[list[tuple[float, float]], bool, bool, bool]], raw_endpoint_count: int, ) -> tuple[ list[tuple[list[tuple[float, float]], bool, bool, bool]], list[tuple[float, float]], int, dict[str, object], ]: """Keep the widest direct backbone for an over-connected flood grid. Synthetic flood grids can have a very long U-shaped graph diameter. Choosing the endpoint pair with the widest straight-line span (then the shorter graph path) avoids that artificial U while preserving an edge-to-edge backbone. The shortest-path formulation also deterministically resolves the occasional one-cycle artifact produced by grid thinning. """ adjacency, _ = skeleton_chain_graph(chains) endpoints = sorted(node for node, edges in adjacency.items() if len(edges) == 1) if raw_endpoint_count <= 4 or len(endpoints) <= 4: active_chains, junction_points, endpoint_count = active_skeleton_chains(chains) return active_chains, junction_points, endpoint_count, { "pruned": False, "diameter_weight": None, "tributary_weights": [], } path_cache = { endpoint: weighted_skeleton_paths(adjacency, endpoint) for endpoint in endpoints } diameter_start: tuple[float, float] | None = None diameter_end: tuple[float, float] | None = None diameter_span = -1.0 diameter_weight = math.inf for start_index, start in enumerate(endpoints): distances, _ = path_cache[start] for end in endpoints[start_index + 1 :]: distance = distances.get(end) if distance is None: continue span = math.hypot(end[0] - start[0], end[1] - start[1]) pair = (start, end) best_pair = (diameter_start, diameter_end) if span > diameter_span + 1e-9 or ( abs(span - diameter_span) <= 1e-9 and ( distance < diameter_weight - 1e-9 or ( abs(distance - diameter_weight) <= 1e-9 and (diameter_start is None or pair < best_pair) ) ) ): diameter_start, diameter_end = pair diameter_span = span diameter_weight = distance if diameter_start is None or diameter_end is None: active_chains, junction_points, endpoint_count = active_skeleton_chains(chains) return active_chains, junction_points, endpoint_count, { "pruned": False, "diameter_weight": None, "tributary_weights": [], } _, diameter_parents = path_cache[diameter_start] diameter_nodes, diameter_edges = skeleton_path_to( diameter_parents, diameter_start, diameter_end, ) diameter_node_set = set(diameter_nodes) kept_indices = set(diameter_edges) tributaries: list[tuple[float, tuple[float, float], list[int]]] = [] for endpoint in endpoints: if endpoint in diameter_node_set: continue distances, parents = path_cache[endpoint] reachable = [node for node in diameter_node_set if node in distances] if not reachable: continue attachment = min(reachable, key=lambda node: (distances[node], node)) _, branch_edges = skeleton_path_to(parents, endpoint, attachment) if branch_edges: tributaries.append((distances[attachment], endpoint, branch_edges)) tributaries.sort(key=lambda branch: (-branch[0], branch[1], tuple(branch[2]))) # Extra terminal branches in the synthetic 40-43 flood grids still read as # a comb/U-shaped canal after cycle pruning. The direct backbone already # spans the encounter, so omit optional tributaries from the visual water. selected_tributaries = [] for _, _, branch_edges in selected_tributaries: kept_indices.update(branch_edges) active_chains, junction_points, endpoint_count = active_skeleton_chains( chains, kept_indices, ) return active_chains, junction_points, endpoint_count, { "pruned": len(kept_indices) < len(chains), "diameter_weight": diameter_weight, "tributary_weights": [weight for weight, _, _ in selected_tributaries], } def sparse_branched_river_layout( component: list[tuple[int, int]], component_width: int, component_height: int, thick_floodplain: bool, ) -> dict[str, object] | None: """Return a medial layout only for sparse, band-like complex components.""" if not thick_floodplain: return None major = max(component_width, component_height) minor = min(component_width, component_height) density = len(component) / max(1, component_width * component_height) if density > 0.55: return None distances = component_grid_distances(component) max_radius = max(distances.values(), default=0) if max_radius > max(6, round(major * 0.14)): return None skeleton = thin_grid_component(component) raw_chains, raw_junction_points, raw_endpoint_count = trace_skeleton_chains( skeleton, component, ) aspect = major / max(1, minor) raw_open_chains = [points for points, _, _, closed in raw_chains if not closed] raw_tortuosity = max( (polyline_tortuosity(points) for points in raw_open_chains), default=math.inf, ) complex_topology = ( aspect < 1.45 or raw_endpoint_count != 2 or bool(raw_junction_points) or len(raw_chains) != 1 or raw_tortuosity > 1.22 ) if not complex_topology: return None chains, junction_points, endpoint_count, pruning = prune_skeleton_branches( raw_chains, raw_endpoint_count, ) open_chains = [points for points, _, _, closed in chains if not closed] tortuosity = max((polyline_tortuosity(points) for points in open_chains), default=math.inf) return { "chains": chains, "density": density, "distances": distances, "endpoint_count": endpoint_count, "junction_points": junction_points, "max_radius": max_radius, "raw_chain_count": len(raw_chains), "raw_endpoint_count": raw_endpoint_count, "raw_junction_count": len(raw_junction_points), "skeleton": skeleton, "tortuosity": tortuosity, **pruning, } def chaikin_open(points: list[tuple[float, float]], rounds: int = 2) -> list[tuple[float, float]]: result = points for _ in range(rounds): if len(result) < 3: break smoothed = [result[0]] for first, second in zip(result, result[1:]): smoothed.extend( ( (first[0] * 0.75 + second[0] * 0.25, first[1] * 0.75 + second[1] * 0.25), (first[0] * 0.25 + second[0] * 0.75, first[1] * 0.25 + second[1] * 0.75), ) ) smoothed.append(result[-1]) result = smoothed return result def organic_skeleton_chain( points: list[tuple[float, float]], distances: dict[tuple[int, int], int], width_tiles: float, rng: random.Random, ) -> list[tuple[float, float]]: anchors = [points[0]] anchors.extend(points[index] for index in range(3, len(points) - 1, 3)) if points[-1] != anchors[-1]: anchors.append(points[-1]) if len(anchors) <= 2: return anchors organic = [anchors[0]] wander = 0.0 for index in range(1, len(anchors) - 1): previous, point, following = anchors[index - 1], anchors[index], anchors[index + 1] tangent_x = following[0] - previous[0] tangent_y = following[1] - previous[1] tangent_length = max(1e-6, math.hypot(tangent_x, tangent_y)) normal_x, normal_y = -tangent_y / tangent_length, tangent_x / tangent_length cell = (math.floor(point[0]), math.floor(point[1])) clearance = max(0.0, distances.get(cell, 1) - width_tiles / 2 - 0.5) # Broad synthetic bands need a visible low-frequency shoreline drift; # sub-tile noise alone still reads as a ruler-straight L/T junction at # overview scale. Clearance keeps the curve safely inside gameplay's # logical river cells before the final mask clip. limit = min(1.35, clearance * 0.70) wander = max(-limit, min(limit, wander * 0.72 + rng.uniform(-0.70, 0.70))) organic.append((point[0] + normal_x * wander, point[1] + normal_y * wander)) organic.append(anchors[-1]) return chaikin_open(organic) def inset_polyline_endpoint( points: list[tuple[float, float]], distance: float, from_start: bool, ) -> tuple[float, float]: if len(points) < 2 or distance <= 0: return points[0] if from_start else points[-1] total_length = polyline_length(points) remaining = min(distance, total_length * 0.45) ordered = points if from_start else list(reversed(points)) for first, second in zip(ordered, ordered[1:]): segment_length = math.hypot(second[0] - first[0], second[1] - first[1]) if segment_length <= 1e-9: continue if remaining <= segment_length: fraction = remaining / segment_length return ( first[0] + (second[0] - first[0]) * fraction, first[1] + (second[1] - first[1]) * fraction, ) remaining -= segment_length return ordered[-1] def endpoint_touches_map_edge( point: tuple[float, float], grid_width: int, grid_height: int, ) -> bool: epsilon = 1e-6 return ( point[0] <= epsilon or point[0] >= grid_width - epsilon or point[1] <= epsilon or point[1] >= grid_height - epsilon ) def draw_medial_river_component( channel_draw: ImageDraw.ImageDraw, component: list[tuple[int, int]], layout: dict[str, object], tile: int, seed: int, grid_width: int, grid_height: int, ) -> None: skeleton = layout["skeleton"] distances = layout["distances"] chains = layout["chains"] junction_points = layout["junction_points"] assert isinstance(skeleton, set) assert isinstance(distances, dict) assert isinstance(chains, list) assert isinstance(junction_points, list) radii = sorted(distances[point] for point in skeleton) middle = len(radii) // 2 median_radius = ( radii[middle] if len(radii) % 2 else (radii[middle - 1] + radii[middle]) / 2 ) width_tiles = max(2.0, min(2.8, median_radius * 2 * 0.34)) channel_width = round(width_tiles * tile) component_salt = len(component) * 0x45D9F3B for x, y in component: component_salt ^= coordinate_hash(x, y, 0, 0xA24BAED5) min_x = min(x for x, _ in component) min_y = min(y for _, y in component) rng = random.Random(coordinate_hash(min_x, min_y, seed, component_salt)) radius = channel_width / 2 for points, start_is_endpoint, end_is_endpoint, closed in chains: if closed: organic = points[::3] or points if organic[-1] != organic[0]: organic = organic + [organic[0]] else: organic = organic_skeleton_chain(points, distances, width_tiles, rng) original = organic organic = list(organic) # Leave enough logical-bank clearance for the full round cap. # Narrow five-cell flood bands otherwise clip the cap back into a # flat plateau even though wider late-game channels look correct. endpoint_inset = width_tiles * 0.72 if start_is_endpoint and not endpoint_touches_map_edge( points[0], grid_width, grid_height, ): organic[0] = inset_polyline_endpoint(original, endpoint_inset, True) if end_is_endpoint and not endpoint_touches_map_edge( points[-1], grid_width, grid_height, ): organic[-1] = inset_polyline_endpoint(original, endpoint_inset, False) pixel_points = [(round(x * tile), round(y * tile)) for x, y in organic] if len(pixel_points) < 2: continue channel_draw.line(pixel_points, fill=255, width=channel_width, joint="curve") if start_is_endpoint: x, y = pixel_points[0] channel_draw.ellipse((x - radius, y - radius, x + radius, y + radius), fill=255) if end_is_endpoint: x, y = pixel_points[-1] channel_draw.ellipse((x - radius, y - radius, x + radius, y + radius), fill=255) for x, y in junction_points: pixel_x, pixel_y = round(x * tile), round(y * tile) channel_draw.ellipse( (pixel_x - radius, pixel_y - radius, pixel_x + radius, pixel_y + radius), fill=255, ) def river_masks(terrain: list[list[str]], tile: int, seed: int) -> tuple[Image.Image, Image.Image]: """Return logical floodplain and visual water masks. Normal one- or two-cell rivers keep their full footprint. Very thick late-game river blocks become either a simple axial channel or a medial skeleton that preserves sparse L/T/U-shaped branch topology. """ logical = region_mask(terrain, "river", tile, seed) visual = logical.copy() channels = Image.new("L", logical.size, 0) channel_draw = ImageDraw.Draw(channels) visual_draw = ImageDraw.Draw(visual) rng = random.Random(seed * 227) for component in terrain_components(terrain, "river"): xs = [x for x, _ in component] ys = [y for _, y in component] min_x, max_x = min(xs), max(xs) min_y, max_y = min(ys), max(ys) component_width = max_x - min_x + 1 component_height = max_y - min_y + 1 major = max(component_width, component_height) minor = min(component_width, component_height) aspect = major / max(1, minor) thick_floodplain = minor >= 5 and len(component) >= major * 3 elongated_channel = aspect >= 1.45 and major >= 4 if not thick_floodplain and not elongated_channel: continue medial_layout = sparse_branched_river_layout( component, component_width, component_height, thick_floodplain, ) component_density = len(component) / max(1, component_width * component_height) compact_lake = aspect < 1.45 and medial_layout is None and component_density >= 0.55 # Sparse band networks use the topology-preserving medial layout. A # genuinely compact near-square water body can still read as a lake; # irregular sparse shapes must never fall back to the bbox ellipse. if aspect < 1.45 and medial_layout is None and not compact_lake: continue clear_padding = max(4, tile // 3) for x, y in component: visual_draw.rectangle( ( x * tile - clear_padding, y * tile - clear_padding, (x + 1) * tile + clear_padding, (y + 1) * tile + clear_padding, ), fill=0, ) if medial_layout is not None: draw_medial_river_component( channel_draw, component, medial_layout, tile, seed, len(terrain[0]), len(terrain), ) continue if compact_lake: center_x = (min_x + max_x + 1) * tile / 2 center_y = (min_y + max_y + 1) * tile / 2 radius_x = component_width * tile * rng.uniform(0.25, 0.37) radius_y = component_height * tile * rng.uniform(0.25, 0.37) channel_draw.ellipse( (center_x - radius_x, center_y - radius_y, center_x + radius_x, center_y + radius_y), fill=255, ) continue horizontal = component_width >= component_height cells_by_axis: dict[int, list[int]] = {} for x, y in component: axis, cross = (x, y) if horizontal else (y, x) cells_by_axis.setdefault(axis, []).append(cross) start_axis = min(cells_by_axis) end_axis = max(cells_by_axis) points: list[tuple[int, int]] = [] wander = 0.0 axis_values = list(range(start_axis, end_axis + 1, 3)) if not axis_values or axis_values[-1] != end_axis: axis_values.append(end_axis) for axis in axis_values: available = cells_by_axis.get(axis) if not available: nearest_axis = min(cells_by_axis, key=lambda candidate: abs(candidate - axis)) available = cells_by_axis[nearest_axis] low, high = min(available), max(available) center_cross = (low + high + 1) * tile / 2 span = max(tile, (high - low + 1) * tile) wander = wander * 0.62 + rng.uniform(-tile * 0.72, tile * 0.72) wander = max(-span * 0.24, min(span * 0.24, wander)) axis_pixel = axis * tile + tile // 2 cross_pixel = round(center_cross + wander) points.append((axis_pixel, cross_pixel) if horizontal else (cross_pixel, axis_pixel)) if len(points) == 1: points.append((points[0][0] + (tile if horizontal else 0), points[0][1] + (0 if horizontal else tile))) if thick_floodplain: channel_width = max(round(tile * 1.35), min(round(tile * 2.8), round(minor * tile * 0.28))) else: channel_width = max(round(tile * 0.88), min(round(tile * 2.25), round(minor * tile * 0.88))) # Normalize axial endpoints to the logical floodplain edge. Internal # ends then move roughly one cap radius inward so clipping cannot turn # the round cap into a flat wall; map-edge ends stay flush off-map. if horizontal: points[0] = (start_axis * tile, points[0][1]) points[-1] = ((end_axis + 1) * tile, points[-1][1]) start_off_map = start_axis == 0 end_off_map = end_axis == len(terrain[0]) - 1 else: points[0] = (points[0][0], start_axis * tile) points[-1] = (points[-1][0], (end_axis + 1) * tile) start_off_map = start_axis == 0 end_off_map = end_axis == len(terrain) - 1 original_points = list(points) endpoint_inset = channel_width * 0.45 if not start_off_map: points[0] = inset_polyline_endpoint(original_points, endpoint_inset, True) if not end_off_map: points[-1] = inset_polyline_endpoint(original_points, endpoint_inset, False) points = [(round(x), round(y)) for x, y in points] channel_draw.line(points, fill=255, width=channel_width, joint="curve") cap_radius = channel_width / 2 for cap_x, cap_y in (points[0], points[-1]): channel_draw.ellipse( ( cap_x - cap_radius, cap_y - cap_radius, cap_x + cap_radius, cap_y + cap_radius, ), fill=255, ) softened_channels = channels.filter(ImageFilter.GaussianBlur(max(2, tile * 0.055))) clipped_channels = ImageChops.darker(softened_channels, logical) combined = ImageChops.lighter(visual, clipped_channels) visual.close() channels.close() softened_channels.close() clipped_channels.close() return logical, combined def selected_road_edges(terrain: list[list[str]], seed: int) -> list[tuple[tuple[int, int], tuple[int, int]]]: rows = len(terrain) columns = len(terrain[0]) def road_degree(x: int, y: int) -> int: return sum( 1 for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)) if 0 <= y + dy < rows and 0 <= x + dx < columns and terrain[y + dy][x + dx] == "road" ) def lane_selected(index: int, salt: int) -> bool: value = (index * 1_103_515_245 + seed * 12_345 + salt * 2_654_435_761) & 0xFFFFFFFF return value % 100 < 26 edges: list[tuple[tuple[int, int], tuple[int, int]]] = [] for y, row in enumerate(terrain): for x, kind in enumerate(row): if kind != "road": continue current_degree = road_degree(x, y) for dx, dy in ((1, 0), (0, 1)): nx, ny = x + dx, y + dy if not (0 <= ny < rows and 0 <= nx < columns): continue neighbor_kind = terrain[ny][nx] if neighbor_kind in STRUCTURE_KINDS: edges.append(((x, y), (nx, ny))) continue if neighbor_kind != "road": continue neighbor_degree = road_degree(nx, ny) narrow_route = current_degree <= 2 or neighbor_degree <= 2 broad_lane = lane_selected(y, 17) if dx else lane_selected(x, 31) edge_variation = ((x * 928_371 + y * 689_287 + seed * 97) % 100) < 12 if narrow_route or broad_lane or edge_variation: edges.append(((x, y), (nx, ny))) return edges def road_paths(terrain: list[list[str]], tile: int, seed: int) -> list[list[tuple[int, int]]]: rng = random.Random(seed * 239) broad_cells: set[tuple[int, int]] = set() paths: list[list[tuple[int, int]]] = [] def component_path( component: list[tuple[int, int]], horizontal: bool, fraction: float, salt: int, ) -> list[tuple[int, int]]: cells_by_axis: dict[int, list[int]] = {} for x, y in component: axis, cross = (x, y) if horizontal else (y, x) cells_by_axis.setdefault(axis, []).append(cross) start_axis, end_axis = min(cells_by_axis), max(cells_by_axis) axes = list(range(start_axis, end_axis + 1, 3)) if axes[-1] != end_axis: axes.append(end_axis) local_rng = random.Random(seed * 257 + salt) points: list[tuple[int, int]] = [] wander = 0.0 for axis in axes: available = cells_by_axis.get(axis) if not available: nearest_axis = min(cells_by_axis, key=lambda candidate: abs(candidate - axis)) available = cells_by_axis[nearest_axis] ordered = sorted(available) base = ordered[min(len(ordered) - 1, round((len(ordered) - 1) * fraction))] wander = wander * 0.58 + local_rng.uniform(-0.55, 0.55) cross = max(min(ordered) + 0.25, min(max(ordered) + 0.75, base + 0.5 + wander)) axis_pixel = axis * tile + tile // 2 cross_pixel = round(cross * tile) points.append((axis_pixel, cross_pixel) if horizontal else (cross_pixel, axis_pixel)) return points for component_index, component in enumerate(terrain_components(terrain, "road")): xs = [x for x, _ in component] ys = [y for _, y in component] width = max(xs) - min(xs) + 1 height = max(ys) - min(ys) + 1 density = len(component) / max(1, width * height) broad = len(component) >= 90 and min(width, height) >= 6 and (density >= 0.24 or len(component) >= 500) if not broad: continue broad_cells.update(component) if width >= height: fractions = (0.28, 0.62, 0.82) if len(component) >= 1200 else (0.34, 0.7) for offset, fraction in enumerate(fractions): paths.append(component_path(component, True, fraction, component_index * 17 + offset)) paths.append(component_path(component, False, 0.52, component_index * 17 + 11)) else: fractions = (0.28, 0.62, 0.82) if len(component) >= 1200 else (0.34, 0.7) for offset, fraction in enumerate(fractions): paths.append(component_path(component, False, fraction, component_index * 17 + offset)) paths.append(component_path(component, True, 0.52, component_index * 17 + 11)) for (x, y), (nx, ny) in selected_road_edges(terrain, seed): if (x, y) in broad_cells and (nx, ny) in broad_cells: continue x1, y1 = x * tile + tile // 2, y * tile + tile // 2 x2, y2 = nx * tile + tile // 2, ny * tile + tile // 2 mid = ( round((x1 + x2) / 2 + rng.randint(-tile // 7, tile // 7)), round((y1 + y2) / 2 + rng.randint(-tile // 7, tile // 7)), ) paths.append([(x1, y1), mid, (x2, y2)]) return [path for path in paths if len(path) >= 2] def road_mask(terrain: list[list[str]], tile: int, seed: int) -> Image.Image: width = len(terrain[0]) * tile height = len(terrain) * tile mask = Image.new("L", (width, height), 0) draw = ImageDraw.Draw(mask) road_width = max(11, round(tile * 0.24)) for path in road_paths(terrain, tile, seed): draw.line(path, fill=255, width=road_width, joint="curve") return mask.filter(ImageFilter.GaussianBlur(max(1.0, tile * 0.055))) def shifted_mask(mask: Image.Image, dx: int, dy: int) -> Image.Image: shifted = Image.new("L", mask.size, 0) shifted.paste(mask, (dx, dy)) return shifted def composite_texture(canvas: Image.Image, texture: Image.Image, mask: Image.Image) -> Image.Image: canvas.paste(texture, (0, 0), mask) texture.close() return canvas def apply_tint(canvas: Image.Image, mask: Image.Image, color: tuple[int, int, int], strength: float) -> Image.Image: tint_mask = mask.point(lambda value: round(value * strength)) canvas.paste(color, (0, 0, canvas.width, canvas.height), tint_mask) tint_mask.close() return canvas def paste_region_features( canvas: Image.Image, source: Image.Image, terrain: list[list[str]], kind: str, region: Image.Image, tile: int, seed: int, ) -> Image.Image: """Scatter large feathered material crops instead of one crop per cell.""" rng = random.Random(seed * 181 + len(kind) * 37) cells = [(x, y) for y, row in enumerate(terrain) for x, value in enumerate(row) if value == kind] if not cells: return canvas density = { "forest": (31, 43), "hill": (12, 34), "cliff": (17, 48), }[kind] multipliers = { "forest": (2.15, 2.65, 3.15), "hill": (2.75, 3.35, 3.95), "cliff": (2.35, 2.9, 3.45), }[kind] variants: list[tuple[Image.Image, Image.Image]] = [] for multiplier in multipliers: side = max(tile, round(tile * multiplier)) for index in range(6): crop_side = round(source.width * rng.uniform(0.36, 0.62)) left = rng.randint(0, max(0, source.width - crop_side)) top = rng.randint(0, max(0, source.height - crop_side)) crop = source.crop((left, top, left + crop_side, top + crop_side)) feature_mask = None if kind == "forest": feature_mask = ImageOps.grayscale(crop).point( lambda value: max(0, min(255, (82 - value) * 5)) ) feature_mask = ImageOps.fit(feature_mask, (side, side), method=Image.Resampling.LANCZOS) variant = ImageOps.fit(crop, (side, side), method=Image.Resampling.LANCZOS) crop.close() if index % 3 == 1: variant = ImageOps.mirror(variant) if feature_mask is not None: feature_mask = ImageOps.mirror(feature_mask) elif index % 3 == 2: variant = variant.transpose(Image.Transpose.ROTATE_180) if feature_mask is not None: feature_mask = feature_mask.transpose(Image.Transpose.ROTATE_180) variant = ImageEnhance.Brightness(variant).enhance(rng.uniform(0.94, 1.05)) soft_mask = feathered_patch_mask(side, 242 if kind == "cliff" else 232) if feature_mask is not None: feature_mask = feature_mask.filter(ImageFilter.GaussianBlur(max(1.5, tile * 0.035))) combined_mask = ImageChops.multiply(soft_mask, feature_mask) soft_mask.close() feature_mask.close() soft_mask = combined_mask variants.append((variant, soft_mask)) candidates: list[tuple[int, int, int]] = [] for x, y in cells: boundary = any(not is_kind(terrain, x + dx, y + dy, kind) for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1))) threshold = density[1] if boundary else density[0] value = coordinate_hash(x, y, seed, 0xA511E9B3) if value % 100 < threshold: candidates.append((value, x, y)) candidates.sort() exclusion = {"forest": 1, "hill": 2, "cliff": 2}[kind] selected: list[tuple[int, int]] = [] selected_set: set[tuple[int, int]] = set() for _, x, y in candidates: if any( (x + dx, y + dy) in selected_set for dx in range(-exclusion, exclusion + 1) for dy in range(-exclusion, exclusion + 1) if dx * dx + dy * dy <= exclusion * exclusion ): continue selected.append((x, y)) selected_set.add((x, y)) if not selected: selected.append(cells[len(cells) // 2]) max_jitter = max(2, round(tile * 0.34)) for x, y in selected: index = coordinate_hash(x, y, seed, 0x63D83595) % len(variants) variant, soft_mask = variants[index] side = variant.width left = x * tile + (tile - side) // 2 + rng.randint(-max_jitter, max_jitter) top = y * tile + (tile - side) // 2 + rng.randint(-max_jitter, max_jitter) clip = region.crop((left, top, left + side, top + side)) bleed_size = min(side - 1, max(5, round(tile * 0.72)) | 1) expanded_clip = clip.filter(ImageFilter.MaxFilter(bleed_size)) combined = ImageChops.multiply(soft_mask, expanded_clip) canvas.paste(variant, (left, top), combined) clip.close() expanded_clip.close() combined.close() for variant, soft_mask in variants: variant.close() soft_mask.close() return canvas def structure_variant_mask(variant: Image.Image, tile: int) -> Image.Image: side = variant.width blurred = variant.filter(ImageFilter.GaussianBlur(max(5, round(side * 0.085)))) detail = ImageOps.grayscale(ImageChops.difference(variant, blurred)) blurred.close() detail = detail.point(lambda value: max(0, min(255, (value - 5) * 18))) dilation = min(side - 1, max(7, round(side * 0.13)) | 1) detail = detail.filter(ImageFilter.MaxFilter(dilation)) detail = detail.filter(ImageFilter.GaussianBlur(max(2, round(side * 0.035)))) radial = Image.new("L", (side, side), 0) radial_draw = ImageDraw.Draw(radial) inset = max(2, round(tile * 0.08)) radial_draw.ellipse((inset, inset, side - inset, side - inset), fill=255) radial = radial.filter(ImageFilter.GaussianBlur(max(8, round(side * 0.11)))) ground_halo = radial.point(lambda value: round(value * 0.16)) foreground = ImageChops.multiply(detail, radial) result = ImageChops.lighter(foreground, ground_halo) detail.close() radial.close() ground_halo.close() foreground.close() return result def add_region(canvas: Image.Image, sources: dict[str, Image.Image], terrain: list[list[str]], kind: str, tile: int, seed: int) -> Image.Image: mask = region_mask(terrain, kind, tile, seed) if mask.getbbox() is None: mask.close() return canvas if kind in {"forest", "hill", "cliff"}: shifted = shifted_mask(mask, max(2, tile // 18), max(3, tile // 14)) shadow_mask = ImageChops.subtract(shifted, mask).filter(ImageFilter.GaussianBlur(max(2, tile // 11))) shadow = Image.new("RGB", canvas.size, (24, 25, 18)) shadow_strength = shadow_mask.point(lambda value: round(value * (0.38 if kind == "cliff" else 0.24))) shadowed = Image.composite(shadow, canvas, shadow_strength) canvas.close() canvas = shadowed shadow.close() shifted.close() shadow_mask.close() shadow_strength.close() tint, strength = { "forest": ((45, 60, 34), 0.035), "hill": ((113, 96, 63), 0.02), "cliff": ((111, 101, 79), 0.08), }[kind] canvas = apply_tint(canvas, mask, tint, strength) canvas = paste_region_features(canvas, sources[kind], terrain, kind, mask, tile, seed + 17) mask.close() return canvas def add_river(canvas: Image.Image, sources: dict[str, Image.Image], terrain: list[list[str]], tile: int, seed: int) -> Image.Image: logical_mask, mask = river_masks(terrain, tile, seed) if mask.getbbox() is None: logical_mask.close() mask.close() return canvas canvas = apply_tint(canvas, logical_mask, (82, 86, 62), 0.075) bank_size = max(5, round(tile * 0.32)) | 1 expanded = mask.filter(ImageFilter.MaxFilter(bank_size)) bank = ImageChops.subtract(expanded, mask).filter(ImageFilter.GaussianBlur(max(1, tile // 24))) bank_layer = Image.new("RGB", canvas.size, (67, 64, 43)) bank_strength = bank.point(lambda value: round(value * 0.78)) banked = Image.composite(bank_layer, canvas, bank_strength) canvas.close() canvas = banked bank_layer.close() bank_strength.close() texture = fill_texture(sources["river"], canvas.size, tile, "river", seed + 29) water_tint = Image.new("RGB", canvas.size, (43, 71, 73)) tinted_texture = Image.blend(texture, water_tint, 0.22) texture.close() water_tint.close() texture = tinted_texture canvas = composite_texture(canvas, texture, mask) expanded.close() bank.close() logical_mask.close() mask.close() return canvas def add_roads(canvas: Image.Image, sources: dict[str, Image.Image], terrain: list[list[str]], tile: int, seed: int) -> Image.Image: mask = road_mask(terrain, tile, seed) if mask.getbbox() is None: mask.close() return canvas shadow_mask = shifted_mask(mask, max(2, tile // 25), max(2, tile // 20)).filter(ImageFilter.GaussianBlur(max(2, tile // 14))) shadow_layer = Image.new("RGB", canvas.size, (61, 48, 32)) shadow_strength = shadow_mask.point(lambda value: round(value * 0.20)) shadowed = Image.composite(shadow_layer, canvas, shadow_strength) canvas.close() canvas = shadowed shadow_layer.close() shadow_mask.close() shadow_strength.close() packed_earth = Image.new("RGB", canvas.size, (154, 127, 84)) packed_route = Image.blend(canvas, packed_earth, 0.38) routed = Image.composite(packed_route, canvas, mask) canvas.close() canvas = routed packed_earth.close() packed_route.close() mask.close() return canvas def add_structures(canvas: Image.Image, sources: dict[str, Image.Image], terrain: list[list[str]], tile: int, seed: int) -> Image.Image: rng = random.Random(seed * 149) for kind in ("village", "fort", "camp"): cells = {(x, y) for y, row in enumerate(terrain) for x, terrain_kind in enumerate(row) if terrain_kind == kind} if not cells: continue components: list[list[tuple[int, int]]] = [] remaining = set(cells) while remaining: start = remaining.pop() component = [start] stack = [start] while stack: x, y = stack.pop() for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): neighbor = (x + dx, y + dy) if neighbor in remaining: remaining.remove(neighbor) component.append(neighbor) stack.append(neighbor) components.append(component) selected_cells: set[tuple[int, int]] = set() for component in components: center_x = sum(x for x, _ in component) / len(component) center_y = sum(y for _, y in component) / len(component) representative = min(component, key=lambda point: (point[0] - center_x) ** 2 + (point[1] - center_y) ** 2) selected_cells.add(representative) if len(component) <= 3: selected_cells.update(component) else: target = min( {"village": 6, "camp": 6, "fort": 8}[kind], max(2, 1 + round(math.sqrt(len(component)) / (2.5 if kind == "fort" else 2.1))), ) minimum_distance = 3.1 if kind == "fort" else 2.35 candidates = component[:] rng.shuffle(candidates) component_selected = [representative] for candidate in candidates: if len(component_selected) >= target: break if all( (candidate[0] - point[0]) ** 2 + (candidate[1] - point[1]) ** 2 >= minimum_distance**2 for point in component_selected ): component_selected.append(candidate) selected_cells.update(component_selected) source = sources[kind] half_w, half_h = source.width // 2, source.height // 2 quadrants = [ source.crop((0, 0, half_w, half_h)), source.crop((half_w, 0, source.width, half_h)), source.crop((0, half_h, half_w, source.height)), source.crop((half_w, half_h, source.width, source.height)), ] variant_size = max( tile, round(tile * {"village": 1.9, "camp": 2.0, "fort": 2.35}[kind]), ) variants = [ImageOps.fit(variant, (variant_size, variant_size), method=Image.Resampling.LANCZOS) for variant in quadrants] for variant in quadrants: variant.close() variant_masks = [structure_variant_mask(variant, tile) for variant in variants] for x, y in sorted(selected_cells, key=lambda point: (point[1], point[0])): index = coordinate_hash(x, y, seed, 0xC2B2AE35) % len(variants) variant = variants[index] left = x * tile + (tile - variant_size) // 2 + rng.randint(-max(1, tile // 24), max(1, tile // 24)) top = y * tile + (tile - variant_size) // 2 + rng.randint(-max(1, tile // 24), max(1, tile // 24)) canvas.paste(variant, (left, top), variant_masks[index]) for variant in variants: variant.close() for variant_mask in variant_masks: variant_mask.close() return canvas def add_route_ruts(canvas: Image.Image, terrain: list[list[str]], tile: int, seed: int) -> Image.Image: overlay = Image.new("RGBA", canvas.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(overlay) for path in road_paths(terrain, tile, seed): horizontal = abs(path[-1][0] - path[0][0]) >= abs(path[-1][1] - path[0][1]) normal_x, normal_y = (0, 1) if horizontal else (1, 0) offset = max(3, tile // 11) for sign in (-1, 1): jx = normal_x * offset * sign jy = normal_y * offset * sign draw.line( [(x + jx, y + jy) for x, y in path], fill=(70, 52, 34, 58), width=max(1, tile // 34), joint="curve", ) composited = Image.alpha_composite(canvas.convert("RGBA"), overlay.filter(ImageFilter.GaussianBlur(0.55))).convert("RGB") canvas.close() canvas = composited overlay.close() return canvas def theme_settings(number: int) -> tuple[tuple[int, int, int], float, float, float, float]: if number <= 4: return (120, 105, 66), 0.035, 0.97, 0.94, 0.98 if number <= 6: return (126, 108, 77), 0.055, 0.96, 0.94, 1.0 if number <= 15: return (74, 96, 76), 0.045, 0.95, 0.93, 0.98 if number <= 19: return (61, 93, 57), 0.06, 0.96, 0.98, 0.98 if number <= 24: return (49, 87, 88), 0.055, 0.95, 0.93, 0.99 if number <= 37: return (125, 100, 69), 0.07, 0.96, 0.9, 1.01 if number <= 44: return (52, 82, 80), 0.065, 0.94, 0.88, 0.98 if number <= 46: return (117, 76, 48), 0.075, 0.93, 0.88, 1.0 if number <= 54: return (57, 87, 50), 0.085, 0.96, 1.03, 1.0 if number <= 63: return (139, 111, 72), 0.08, 0.96, 0.86, 1.01 return (130, 106, 68), 0.09, 0.96, 0.83, 1.01 def add_event_atmosphere(canvas: Image.Image, terrain: list[list[str]], tile: int, number: int, seed: int) -> Image.Image: tint, tint_alpha, brightness, saturation, contrast = theme_settings(number) canvas = Image.blend(canvas, Image.new("RGB", canvas.size, tint), tint_alpha) canvas = ImageEnhance.Color(canvas).enhance(saturation) canvas = ImageEnhance.Contrast(canvas).enhance(contrast) canvas = ImageEnhance.Brightness(canvas).enhance(brightness) if number == 9: canvas = Image.blend(canvas, Image.new("RGB", canvas.size, (25, 36, 48)), 0.28) canvas = ImageEnhance.Brightness(canvas).enhance(0.78) elif number in {22, 46}: rng = random.Random(seed * 211) glow = Image.new("RGBA", canvas.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(glow) candidates = [(x, y) for y, row in enumerate(terrain) for x, kind in enumerate(row) if kind in {"camp", "fort", "forest"}] rng.shuffle(candidates) for x, y in candidates[: max(5, min(18, len(candidates) // 20))]: cx, cy = x * tile + tile // 2, y * tile + tile // 2 radius = rng.randint(tile // 2, tile * 2) draw.ellipse((cx - radius, cy - radius, cx + radius, cy + radius), fill=(198, 77, 24, rng.randint(22, 42))) canvas = Image.alpha_composite(canvas.convert("RGBA"), glow.filter(ImageFilter.GaussianBlur(max(8, tile // 2)))).convert("RGB") glow.close() elif number in {40, 61}: canvas = Image.blend(canvas, Image.new("RGB", canvas.size, (47, 64, 70)), 0.1 if number == 40 else 0.15) canvas = ImageEnhance.Brightness(canvas).enhance(0.91 if number == 61 else 0.96) return canvas def add_finish(canvas: Image.Image, seed: int) -> Image.Image: width, height = canvas.size noise_side = 512 noise_rng = random.Random(seed ^ 0x9E3779B9) noise = Image.frombytes("L", (noise_side, noise_side), noise_rng.randbytes(noise_side * noise_side)) noise = noise.filter(ImageFilter.GaussianBlur(1.1)).resize(canvas.size, Image.Resampling.BILINEAR) noise_color = ImageOps.colorize(noise, (49, 47, 35), (153, 143, 101)) canvas = Image.blend(canvas, noise_color, 0.02) noise.close() noise_color.close() vignette_small = Image.new("L", (512, 512), 0) vdraw = ImageDraw.Draw(vignette_small) vdraw.ellipse((-55, -40, 567, 570), fill=255) vignette_small = vignette_small.filter(ImageFilter.GaussianBlur(52)) vignette = vignette_small.resize((width, height), Image.Resampling.BILINEAR) dark = Image.new("RGB", canvas.size, (22, 22, 17)) edge_mask = ImageOps.invert(vignette).point(lambda value: round(value * 0.11)) canvas = Image.composite(dark, canvas, edge_mask) vignette_small.close() vignette.close() dark.close() edge_mask.close() return canvas def build_map(battle: dict[str, object], sources: dict[str, Image.Image]) -> tuple[Image.Image, int]: number = int(battle["number"]) terrain = battle["terrain"] columns = int(battle["width"]) rows = int(battle["height"]) tile = output_tile_size(columns, rows) size = (columns * tile, rows * tile) seed = 260715 + number * 1009 canvas = fill_texture(sources["plain"], size, tile, "plain", seed) canvas = add_river(canvas, sources, terrain, tile, seed) for index, kind in enumerate(REGION_KINDS): canvas = add_region(canvas, sources, terrain, kind, tile, seed + index * 31) canvas = add_roads(canvas, sources, terrain, tile, seed) canvas = add_route_ruts(canvas, terrain, tile, seed) canvas = add_structures(canvas, sources, terrain, tile, seed) if number in WATER_LAYER_REASSERT_BATTLES: # Reviewed late-game water fields use intentionally generous feature # bleed. Reassert water last so roads and terrain stamps cannot read as # floating tiles inside the channel. canvas = add_river(canvas, sources, terrain, tile, seed) canvas = add_event_atmosphere(canvas, terrain, tile, number, seed) canvas = add_finish(canvas, seed) return canvas, tile def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as file: for chunk in iter(lambda: file.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def write_contact_sheet(battles: list[dict[str, object]], output: Path) -> None: card_width, card_height = 360, 310 columns = 6 rows = math.ceil(len(battles) / columns) sheet = Image.new("RGB", (columns * card_width, rows * card_height), (24, 25, 20)) draw = ImageDraw.Draw(sheet) for index, battle in enumerate(battles): path = ROOT / str(battle["assetPath"]) if not path.is_file(): continue with Image.open(path) as image: preview = ImageOps.contain(image.convert("RGB"), (card_width - 20, card_height - 44), Image.Resampling.LANCZOS) x = (index % columns) * card_width y = (index // columns) * card_height sheet.paste(preview, (x + (card_width - preview.width) // 2, y + 30 + (card_height - 44 - preview.height) // 2)) draw.text((x + 10, y + 8), f"{int(battle['number']):02d} {battle['id']}", fill=(231, 219, 180)) output.parent.mkdir(parents=True, exist_ok=True) sheet.save(output, "JPEG", quality=90, optimize=True) def write_manifest(battles: list[dict[str, object]], output: Path) -> None: rows = [] for battle in battles: path = ROOT / str(battle["assetPath"]) with Image.open(path) as image: width, height = image.size rows.append( { "number": battle["number"], "id": battle["id"], "mapTextureKey": battle["mapTextureKey"], "assetPath": battle["assetPath"], "grid": [battle["width"], battle["height"]], "image": [width, height], "tilePixels": min(width // int(battle["width"]), height // int(battle["height"])), "bytes": path.stat().st_size, "sha256": sha256(path), } ) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps({"pipeline": "battle-map-v2-atlas-compositor", "maps": rows}, indent=2) + "\n", encoding="utf-8") def main() -> None: args = parse_args() selected_numbers = parse_number_set(args.maps) payload = json.loads(args.data.read_text(encoding="utf-8")) battles = payload["battles"] sources = load_sources() for battle in battles: number = int(battle["number"]) if number not in selected_numbers: continue output = ROOT / str(battle["assetPath"]) image, tile = build_map(battle, sources) if image.width > MAX_TEXTURE_EDGE or image.height > MAX_TEXTURE_EDGE: raise RuntimeError(f"{battle['id']}: generated texture {image.size} exceeds {MAX_TEXTURE_EDGE}px budget") output.parent.mkdir(parents=True, exist_ok=True) image.save(output, "WEBP", quality=args.quality, method=4) image.close() print(f"[{number:02d}/66] {battle['id']} -> {output.name} ({int(battle['width']) * tile}x{int(battle['height']) * tile}, {tile}px/tile)") gc.collect() for source in sources.values(): source.close() if not args.skip_artifacts: write_manifest(battles, args.manifest) write_contact_sheet(battles, args.contact_sheet) print(f"Wrote {args.manifest}") print(f"Wrote {args.contact_sheet}") if __name__ == "__main__": main()