#!/usr/bin/env python3 """ Star Map Generator — The Settled Reach Reads tooling/star-map-seed.json, runs the 6-phase generation algorithm, outputs docs/design/star-map.json and 7 d2 files in docs/diagrams/design/. Usage: python3 tooling/generate-star-map.py [--seed-file tooling/star-map-seed.json] Standard library only. No external dependencies. """ from __future__ import annotations import json import math import random import sys import os import argparse from collections import defaultdict, deque from pathlib import Path from typing import Optional # ── Path resolution ──────────────────────────────────────────────────────────── SCRIPT_DIR = Path(__file__).parent REPO_ROOT = SCRIPT_DIR.parent DEFAULT_SEED_FILE = SCRIPT_DIR / "star-map-seed.json" OUTPUT_JSON = REPO_ROOT / "docs" / "design" / "star-map.json" OUTPUT_D2_DIR = REPO_ROOT / "docs" / "diagrams" / "design" SECTORS = [ "core", "north_reach", "west_reach", "south_reach", "east_reach", "deep_frontier", ] SECTOR_LABELS = { "core": "Core", "north_reach": "North Reach", "west_reach": "West Reach", "south_reach": "South Reach", "east_reach": "East Reach", "deep_frontier": "Deep Frontier", } TOPOLOGY_VALUES = [ "dead_end", "spur_end", "through_route", "loop_member", "junction", "hub", ] WAVE_VALUES = [ "wave_1", "wave_2", "wave_3", "wave_4", "wave_5", "unsettled", ] # ── D2 visual constants ──────────────────────────────────────────────────────── D2_BG = "#1a1e24" D2_TXT = "#c8d0e0" D2_ACC = "#c8d8f0" # Node fill color by settlement wave WAVE_FILL = { "wave_1": "#1a2a50", "wave_2": "#2e2800", "wave_3": "#162a1a", "wave_4": "#2e1400", "wave_5": "#2e0a0a", "unsettled": "#1a1e24", } # Node stroke color by settlement wave WAVE_STROKE = { "wave_1": "#3060c0", "wave_2": "#b8a020", "wave_3": "#3a8a50", "wave_4": "#c86010", "wave_5": "#c02020", "unsettled": "#4a5060", } # Cross-sector stub: dimmed STUB_FILL = "#111418" STUB_STROKE = "#3a4050" # Gateway special colors GATEWAY_FILL = "#1a2850" GATEWAY_STROKE = "#5090e0" # Edge color defaults EDGE_COLOR_INTRA = "#4a5a70" EDGE_COLOR_CROSS = "#6a7a40" EDGE_COLOR_GATEWAY = "#5090e0" def d2_node_shape(topology: str) -> str: """Return d2 shape name for topology type.""" if topology == "hub": return "hexagon" elif topology == "junction": return "diamond" elif topology in ("dead_end", "spur_end"): return "rectangle" else: # loop_member, through_route return "oval" # ── Weighted random choice ───────────────────────────────────────────────────── def weighted_choice(rng: random.Random, options: dict) -> str: """Choose from a dict of {value: weight} using the given rng.""" keys = list(options.keys()) weights = [options[k] for k in keys] total = sum(weights) r = rng.random() * total cumulative = 0.0 for k, w in zip(keys, weights): cumulative += w if r <= cumulative: return k return keys[-1] # ── Phase 1: System placement ────────────────────────────────────────────────── def phase1_place_systems(seed_cfg: dict, rng: random.Random) -> list[dict]: """ Create all system nodes with sector, band, wave, star_type assignments. Returns list of node dicts. Gateway is placed first as S-001. """ nodes = [] system_count = seed_cfg["system_count"] sector_dist = seed_cfg["sector_distribution"] wave_by_sector = seed_cfg["settlement_wave_by_sector"] band_by_sector = seed_cfg["geographic_band_by_sector"] star_dist = seed_cfg["star_type_distribution"] gateway_cfg = seed_cfg["gateway"] # Build the Gateway node first gateway_node = { "system_id": "S-001", "system_name": "PLACEHOLDER_GATEWAY", "star_type": "G", "geographic_sector": gateway_cfg["geographic_sector"], "geographic_band": gateway_cfg["geographic_band"], "political_zone": gateway_cfg["political_zone"], "settlement_wave": gateway_cfg["settlement_wave"], "gate_topology": gateway_cfg["gate_topology"], "aperture_count": gateway_cfg["aperture_count"], "gate_connections": 0, # will be set after edge building "_gateway": True, } nodes.append(gateway_node) # Build remaining nodes by sector # The Gateway occupies one core slot adjusted_sector_dist = dict(sector_dist) adjusted_sector_dist["core"] = max(0, sector_dist["core"] - 1) # Expand sector list with correct counts sector_queue = [] for sector, count in adjusted_sector_dist.items(): sector_queue.extend([sector] * count) # Trim or pad to reach total count - 1 (Gateway already placed) target_remaining = system_count - 1 if len(sector_queue) < target_remaining: # Pad with deep_frontier sector_queue.extend(["deep_frontier"] * (target_remaining - len(sector_queue))) elif len(sector_queue) > target_remaining: # Trim from the end (deep_frontier was padded last) sector_queue = sector_queue[:target_remaining] rng.shuffle(sector_queue) counter = 2 # S-001 is Gateway for sector in sector_queue: wave = weighted_choice(rng, wave_by_sector[sector]) band = weighted_choice(rng, band_by_sector[sector]) star_type = weighted_choice(rng, star_dist) node = { "system_id": f"S-{counter:03d}", "system_name": f"PLACEHOLDER_{counter:03d}", "star_type": star_type, "geographic_sector": sector, "geographic_band": band, "political_zone": _assign_political_zone(sector, band, wave, rng), "settlement_wave": wave, "gate_topology": "dead_end", # default; overwritten in Phase 4 "aperture_count": 1, # floor; overwritten in Phase 5 "gate_connections": 0, # set after edges built "_gateway": False, } nodes.append(node) counter += 1 return nodes def _assign_political_zone(sector: str, band: str, wave: str, rng: random.Random) -> str: """ Assign a plausible political zone based on sector/band/wave. Rough heuristic — not setting-perfect but good enough for topology generation. """ if sector == "core": return "institutional_core" if sector == "deep_frontier": return rng.choice(["contested_frontier", "deep_reach_isolate", "deep_reach_isolate"]) if band == "inner": if wave in ("wave_1", "wave_2"): return rng.choice(["institutional_core", "commercial_mid_reach", "commercial_mid_reach"]) elif wave in ("wave_3", "wave_4"): return rng.choice(["commercial_mid_reach", "research_periphery", "contested_frontier"]) else: return rng.choice(["contested_frontier", "commercial_mid_reach"]) else: # outer if wave in ("wave_4", "wave_5", "unsettled"): return rng.choice(["contested_frontier", "deep_reach_isolate"]) else: return rng.choice(["commercial_mid_reach", "research_periphery", "deep_reach_isolate"]) # ── Graph helpers ────────────────────────────────────────────────────────────── def bfs_connected(adj: dict, start: str, allowed: set) -> set: """BFS from start node. Returns set of reachable node IDs (only in allowed set).""" visited = set() queue = deque([start]) while queue: node = queue.popleft() if node in visited: continue visited.add(node) for neighbor in adj.get(node, []): if neighbor not in visited and neighbor in allowed: queue.append(neighbor) return visited def bfs_distances(adj: dict, start: str) -> dict: """BFS from start. Returns dict of {node_id: hop_distance}.""" distances = {start: 0} queue = deque([start]) while queue: node = queue.popleft() for neighbor in adj.get(node, []): if neighbor not in distances: distances[neighbor] = distances[node] + 1 queue.append(neighbor) return distances def build_adjacency(edges: list[list]) -> dict: """Build adjacency dict from edge list.""" adj = defaultdict(list) for a, b in edges: adj[a].append(b) adj[b].append(a) return adj def degree(adj: dict, node_id: str) -> int: return len(adj.get(node_id, [])) def edge_exists(edges_set: set, a: str, b: str) -> bool: return (a, b) in edges_set or (b, a) in edges_set def add_edge(edges: list, edges_set: set, adj: dict, a: str, b: str): """Add edge if it doesn't already exist.""" if a == b: return if edge_exists(edges_set, a, b): return edges.append([a, b]) edges_set.add((a, b)) adj[a].append(b) adj[b].append(a) # ── Phase 2: Spanning tree backbone ─────────────────────────────────────────── def phase2_spanning_tree( nodes: list[dict], seed_cfg: dict, rng: random.Random, ) -> tuple[list[list], set, dict]: """ Build a spanning tree using a modified Prim's algorithm. Returns (edges, edges_set, adj). """ weights = seed_cfg["augmentation_weights"] cross_penalty = weights["cross_sector_weight_penalty"] inner_inner_bonus = weights["inner_to_inner_weight_bonus"] outer_inner_bonus = weights["outer_to_inner_weight_bonus"] node_map = {n["system_id"]: n for n in nodes} all_ids = [n["system_id"] for n in nodes] edges = [] edges_set = set() adj = defaultdict(list) # Start from Gateway (S-001) in_tree = {"S-001"} not_in_tree = set(all_ids) - in_tree while not_in_tree: best_a = None best_b = None best_weight = -999.0 # For efficiency, sample a candidate subset when the tree is large tree_sample = list(in_tree) if len(tree_sample) > 60: tree_sample = rng.sample(tree_sample, 60) not_tree_sample = list(not_in_tree) if len(not_tree_sample) > 60: not_tree_sample = rng.sample(not_tree_sample, 60) for a_id in tree_sample: a = node_map[a_id] for b_id in not_tree_sample: b = node_map[b_id] w = _spanning_tree_weight(a, b, cross_penalty, inner_inner_bonus, outer_inner_bonus, rng, adj) if w > best_weight: best_weight = w best_a = a_id best_b = b_id if best_b is None: # Fallback: pick any unconnected node and connect to nearest tree member b_id = next(iter(not_in_tree)) a_id = rng.choice(list(in_tree)) best_a = a_id best_b = b_id add_edge(edges, edges_set, adj, best_a, best_b) in_tree.add(best_b) not_in_tree.discard(best_b) return edges, edges_set, adj def _spanning_tree_weight( a: dict, b: dict, cross_penalty: float, inner_inner_bonus: float, outer_inner_bonus: float, rng: random.Random, adj: dict, ) -> float: """ Compute connection weight between two nodes for spanning tree. Higher = more likely to connect. Key design goal: produce long chains, not star topologies. We penalise high-degree tree nodes so the tree fans out as a collection of paths rather than a hub-and-spoke web. """ w = rng.random() # base randomness # Penalize cross-sector connections (applied first, on positive base) if a["geographic_sector"] != b["geographic_sector"]: # Allow cross-sector but penalize, except core-to-adjacent (desired) if a["geographic_sector"] == "core" or b["geographic_sector"] == "core": w *= (1.0 - cross_penalty * 0.5) # lighter penalty for core connections elif a["geographic_sector"] == "deep_frontier" or b["geographic_sector"] == "deep_frontier": w *= (1.0 - cross_penalty * 0.8) # heavier penalty for frontier jumps else: w *= (1.0 - cross_penalty) # Bonus for inner-to-inner connections (spine of the network) if a["geographic_band"] == "inner" and b["geographic_band"] == "inner": w += inner_inner_bonus * rng.random() # Bonus for outer connecting to inner (inward-pulling) if (a["geographic_band"] == "outer" and b["geographic_band"] == "inner") or \ (a["geographic_band"] == "inner" and b["geographic_band"] == "outer"): w += outer_inner_bonus * rng.random() # Bonus for same sector connections if a["geographic_sector"] == b["geographic_sector"]: w += 0.2 # Degree penalty on the in-tree node (a) applied last — discourages stars. # A node already at degree 2 in the tree is less attractive as a parent; # this pushes the tree toward chains rather than hub-and-spoke. a_deg = len(adj.get(a["system_id"], [])) if a_deg == 1: w += 0.15 # slight bonus to extend existing chains elif a_deg == 2: w -= 0.25 # mild penalty — prefer not to triple-branch here elif a_deg >= 3: w -= 0.55 # heavy penalty — already a branching node return w # ── Phase 3: Augmentation ────────────────────────────────────────────────────── def phase3_augment( nodes: list[dict], edges: list[list], edges_set: set, adj: dict, seed_cfg: dict, rng: random.Random, ) -> None: """ Run augmentation passes A-D in-place. A: Hub formation B: Loop formation C: Spur extension (dead-ends — no action needed, they're already there) D: Cross-sector bridges """ node_map = {n["system_id"]: n for n in nodes} weights = seed_cfg["augmentation_weights"] hub_targets = seed_cfg["hub_count_targets"] cross_targets = seed_cfg["cross_sector_connection_targets"] hub_min = weights["hub_target_degree_min"] hub_max = weights["hub_target_degree_max"] junc_min = weights["junction_target_degree_min"] junc_max = weights["junction_target_degree_max"] # Pass A: Hub formation # Select hub candidate systems: one Gateway + a tightly controlled count per sector. # hub_count_targets in the seed config are MAXIMUMS, not minimums — we use them # as the exact count to avoid over-producing hubs. hub_candidates = _select_hub_candidates(nodes, seed_cfg, rng) for hub_id in hub_candidates: hub = node_map[hub_id] # Gateway has exactly 4 active connections if hub.get("_gateway"): target_degree = 4 else: # Non-gateway hubs: target degree 5–6 (not the full hub_min/max range # which goes up to 7, producing too many high-degree nodes) target_degree = rng.randint(hub_min, min(hub_max, hub_min + 1)) sector_peers = [ n["system_id"] for n in nodes if n["system_id"] != hub_id and n["geographic_sector"] == hub["geographic_sector"] ] same_sector_inner = [ n["system_id"] for n in nodes if n["system_id"] != hub_id and n["geographic_sector"] == hub["geographic_sector"] and n["geographic_band"] in ("inner", "core") ] candidates = same_sector_inner if same_sector_inner else sector_peers _augment_node_degree(edges, edges_set, adj, hub_id, candidates, target_degree, rng) # Pass A continued: Junction formation — only degree-3 target to avoid # inadvertently inflating future hub counts via cross-sector bridges. junction_candidates = _select_junction_candidates(nodes, hub_candidates, seed_cfg, rng) for junc_id in junction_candidates: target_degree = junc_min # always target the minimum (3) to stay conservative sector_peers = [ n["system_id"] for n in nodes if n["system_id"] != junc_id and n["geographic_sector"] == node_map[junc_id]["geographic_sector"] ] _augment_node_degree(edges, edges_set, adj, junc_id, sector_peers, target_degree, rng) # Pass B: Loop formation — increased target to push loop_member count up loop_target = weights["loop_formation_target_count"] loop_min_path = weights["loop_min_path_length"] loop_max_path = weights["loop_max_path_length"] loops_added = 0 # Try to form loops within sectors for sector in SECTORS: sector_ids = [n["system_id"] for n in nodes if n["geographic_sector"] == sector] if len(sector_ids) < 4: continue sector_id_set = set(sector_ids) attempts = 0 while loops_added < loop_target and attempts < 300: attempts += 1 a_id = rng.choice(sector_ids) # BFS to find nodes at desired path distance dist = bfs_distances(adj, a_id) candidates_for_loop = [ nid for nid, d in dist.items() if loop_min_path <= d <= loop_max_path and nid in sector_id_set and not edge_exists(edges_set, a_id, nid) ] if candidates_for_loop: b_id = rng.choice(candidates_for_loop) add_edge(edges, edges_set, adj, a_id, b_id) loops_added += 1 # Pass C: Leaf reduction — aggressively reduce dead_end count by chaining # leaf nodes to nearby non-leaf nodes (turning leaves into through_routes # and spur_ends, and upgrading degree-2 chains). _reduce_leaves(nodes, edges, edges_set, adj, seed_cfg, rng) # Pass D: Cross-sector bridges # Ensure minimum cross-sector connections per the target config _ensure_cross_sector_bridges(nodes, edges, edges_set, adj, cross_targets, rng) # Pass E: Enforce Gateway connection cap # Gateway should have exactly gateway_cfg["gate_connections"] active edges. # The spanning tree may have created more — remove excess by rerouting. # We do this AFTER all other augmentation to not break the spanning tree. gateway_cfg = seed_cfg["gateway"] gateway_max = gateway_cfg["gate_connections"] # 4 gw_id = "S-001" gw_neighbors = list(adj.get(gw_id, [])) if len(gw_neighbors) > gateway_max: # Remove excess edges — keep the highest-degree neighbors (they're the most connected) sorted_neighbors = sorted( gw_neighbors, key=lambda nid: degree(adj, nid), reverse=True, ) to_keep = set(sorted_neighbors[:gateway_max]) to_remove = [nid for nid in gw_neighbors if nid not in to_keep] for remove_id in to_remove: # Remove from edges list edges[:] = [ e for e in edges if not (set(e) == {gw_id, remove_id}) ] # Remove from edges_set edges_set.discard((gw_id, remove_id)) edges_set.discard((remove_id, gw_id)) # Update adj if remove_id in adj[gw_id]: adj[gw_id].remove(remove_id) if gw_id in adj[remove_id]: adj[remove_id].remove(gw_id) # Reconnect the removed neighbor to a non-gateway core system if needed # (to preserve connectivity) core_systems = [ n["system_id"] for n in nodes if n["geographic_sector"] == "core" and n["system_id"] != gw_id and n["system_id"] != remove_id ] if core_systems: reconnect_target = rng.choice(core_systems) if not edge_exists(edges_set, remove_id, reconnect_target): add_edge(edges, edges_set, adj, remove_id, reconnect_target) def _augment_node_degree( edges: list, edges_set: set, adj: dict, node_id: str, candidates: list, target_degree: int, rng: random.Random, ) -> None: """ Add edges from node_id to candidates until target_degree is reached. Safe against fully-connected candidate lists (terminates when no new edge possible). """ if not candidates: return shuffled = list(candidates) rng.shuffle(shuffled) for cand_id in shuffled: if degree(adj, node_id) >= target_degree: break if not edge_exists(edges_set, node_id, cand_id): add_edge(edges, edges_set, adj, node_id, cand_id) def _select_hub_candidates(nodes: list[dict], seed_cfg: dict, rng: random.Random) -> list[str]: """ Select systems to become hubs. One per sector minimum, plus Gateway. """ hub_targets = seed_cfg["hub_count_targets"] candidates = ["S-001"] # Gateway is always a hub for sector, count in hub_targets.items(): sector_nodes = [ n["system_id"] for n in nodes if n["geographic_sector"] == sector and not n.get("_gateway") and n["geographic_band"] in ("inner", "core") ] if not sector_nodes: sector_nodes = [n["system_id"] for n in nodes if n["geographic_sector"] == sector] selected = rng.sample(sector_nodes, min(count, len(sector_nodes))) candidates.extend(selected) return list(set(candidates)) def _select_junction_candidates( nodes: list[dict], hub_candidates: list[str], seed_cfg: dict, rng: random.Random, ) -> list[str]: """ Select junction candidates: inner-band non-hub systems. We select only half the topology target count here because: - Pass C (leaf reduction) will naturally push many degree-2 nodes to degree 3 as well, producing more junctions organically. - Over-selecting here was one cause of too many hubs (junction nodes at degree 3 get one more edge from cross-sector bridges → degree 4, then classification bumps them to hub tier). """ hub_set = set(hub_candidates) topology_targets = seed_cfg["topology_targets"] total = seed_cfg["system_count"] # Use ~40% of the junction target — the rest come from organic augmentation junction_count = int(total * topology_targets["junction"] * 0.4) candidates = [ n["system_id"] for n in nodes if n["system_id"] not in hub_set and n["geographic_band"] in ("inner", "core") and n["geographic_sector"] != "deep_frontier" ] rng.shuffle(candidates) return candidates[:junction_count] def _reduce_leaves( nodes: list[dict], edges: list[list], edges_set: set, adj: dict, seed_cfg: dict, rng: random.Random, ) -> None: """ Pass C: Leaf reduction. A spanning tree of 300 nodes has ~150 leaves (degree-1 nodes). Without intervention, these remain as dead_ends, which is far above the 20% target. This pass reduces the leaf count to ~38% (about 114 systems), leaving the sculpt pass to bring it to the final 20% target. CRITICAL design constraint: we NEVER connect leaf-to-leaf (which would create a 2-node dangling chain whose edge is a bridge, keeping both as degree-1 effective dead_ends, or worse — if both are in a larger component a direct leaf-to-leaf edge always forms a new cycle via the existing tree path, instantly creating loop_members). Instead we ONLY connect leaves to nearby degree-2 chain nodes: - leaf gains degree 2 (becomes through_route or spur_end candidate) - degree-2 target gains degree 3 (becomes junction) This creates through_routes and junctions organically without cycles. Stopping at ~38% dead_ends gives the sculpt pass a graph that is sparser-than-target (too many dead_ends, too few loops), which is much easier to correct by adding edges than the reverse. """ total = len(nodes) # Stop at ~38% dead_ends — well above the 20% target. # Sculpt will reduce further by adding targeted edges. target_leaf_count = int(total * 0.38) node_map = {n["system_id"]: n for n in nodes} def current_leaves(): return [n["system_id"] for n in nodes if degree(adj, n["system_id"]) == 1] max_rounds = 20 for _round in range(max_rounds): leaves = current_leaves() if len(leaves) <= target_leaf_count: break # Shuffle for variety rng.shuffle(leaves) made_progress = False for leaf_id in leaves: if len(current_leaves()) <= target_leaf_count: break leaf = node_map[leaf_id] # BFS once per leaf dist = bfs_distances(adj, leaf_id) # ONLY connect leaf to a nearby degree-2 same-sector node. # This upgrades the leaf to degree-2 and the target to degree-3 # (junction), creating through_routes — NO cycles formed. sector_d2 = [ nid for nid, d in dist.items() if 2 <= d <= 8 and node_map.get(nid, {}).get("geographic_sector") == leaf["geographic_sector"] and degree(adj, nid) == 2 and not edge_exists(edges_set, leaf_id, nid) ] if sector_d2: # Prefer closer targets; pick from top-5 sector_d2_sorted = sorted(sector_d2, key=lambda nid: dist.get(nid, 9999)) target_node = rng.choice(sector_d2_sorted[:5]) add_edge(edges, edges_set, adj, leaf_id, target_node) made_progress = True continue if not made_progress: # No more degree-2 targets reachable — remaining leaves stay as # dead_ends for the sculpt pass to handle via loop-edge addition. break def _count_cross_sector_edges(edges: list[list], node_map: dict, sector_a: str, sector_b: str) -> int: """Count edges that cross between two specific sectors.""" count = 0 for a_id, b_id in edges: sec_a = node_map[a_id]["geographic_sector"] sec_b = node_map[b_id]["geographic_sector"] if set([sec_a, sec_b]) == set([sector_a, sector_b]): count += 1 return count def _ensure_cross_sector_bridges( nodes: list[dict], edges: list[list], edges_set: set, adj: dict, cross_targets: dict, rng: random.Random, ) -> None: """ Ensure each sector boundary has at least the target number of connections. Adds bridging edges through high-degree systems where possible. """ node_map = {n["system_id"]: n for n in nodes} # Parse cross_targets keys like "core_to_north" boundary_map = { ("core", "north_reach"): cross_targets.get("core_to_north", 2), ("core", "west_reach"): cross_targets.get("core_to_west", 2), ("core", "south_reach"): cross_targets.get("core_to_south", 2), ("core", "east_reach"): cross_targets.get("core_to_east", 2), ("north_reach", "deep_frontier"): cross_targets.get("north_to_deep_frontier", 2), ("west_reach", "deep_frontier"): cross_targets.get("west_to_deep_frontier", 2), ("south_reach", "deep_frontier"): cross_targets.get("south_to_deep_frontier", 2), ("east_reach", "deep_frontier"): cross_targets.get("east_to_deep_frontier", 2), ("north_reach", "west_reach"): cross_targets.get("north_to_west", 1), ("west_reach", "south_reach"): cross_targets.get("west_to_south", 1), ("south_reach", "east_reach"): cross_targets.get("south_to_east", 1), ("east_reach", "north_reach"): cross_targets.get("east_to_north", 1), } for (sec_a, sec_b), target in boundary_map.items(): current = _count_cross_sector_edges(edges, node_map, sec_a, sec_b) needed = target - current if needed <= 0: continue # Pick highest-degree inner nodes from each sector as bridge anchors nodes_a = sorted( [n for n in nodes if n["geographic_sector"] == sec_a], key=lambda n: degree(adj, n["system_id"]), reverse=True, ) nodes_b = sorted( [n for n in nodes if n["geographic_sector"] == sec_b], key=lambda n: degree(adj, n["system_id"]), reverse=True, ) if not nodes_a or not nodes_b: continue added = 0 attempts = 0 while added < needed and attempts < 50: attempts += 1 # Pick a candidate from each sector, weighted toward top of sorted list idx_a = min(int(rng.random() ** 2 * len(nodes_a)), len(nodes_a) - 1) idx_b = min(int(rng.random() ** 2 * len(nodes_b)), len(nodes_b) - 1) a_id = nodes_a[idx_a]["system_id"] b_id = nodes_b[idx_b]["system_id"] if not edge_exists(edges_set, a_id, b_id): add_edge(edges, edges_set, adj, a_id, b_id) added += 1 # ── Phase 4: Topology classification ────────────────────────────────────────── def phase4_classify_topology( nodes: list[dict], edges: list[list], adj: dict, ) -> None: """ Classify each node's gate_topology based on its degree and graph position. Modifies nodes in-place. """ # Detect loop members: nodes that are part of a cycle loop_members = _find_loop_members(nodes, adj) for node in nodes: nid = node["system_id"] d = degree(adj, nid) if d == 0: # Isolated — shouldn't happen after spanning tree node["gate_topology"] = "dead_end" elif d == 1: node["gate_topology"] = "dead_end" elif d == 2: if nid in loop_members: node["gate_topology"] = "loop_member" else: # spur_end: NOT part of a cycle, and at least one neighbour # is a branching node (degree >= 3), meaning this system # hangs off a busier spine. It does NOT need both neighbours # to be high-degree — one busy endpoint is sufficient to # classify a system as a spur rather than a chain link. # through_route: both neighbours are degree <= 2 (pure chain). neighbors = adj.get(nid, []) at_least_one_branching = any( degree(adj, nb) >= 3 for nb in neighbors ) if at_least_one_branching: node["gate_topology"] = "spur_end" else: node["gate_topology"] = "through_route" elif d == 3: if nid in loop_members: node["gate_topology"] = "loop_member" else: node["gate_topology"] = "junction" elif d == 4: node["gate_topology"] = "junction" else: # d >= 5 node["gate_topology"] = "hub" # Override: Gateway is always hub if node.get("_gateway"): node["gate_topology"] = "hub" def _find_loop_members(nodes: list[dict], adj: dict) -> set: """ Find all nodes that participate in at least one cycle. Uses iterative DFS with explicit depth tracking. For each back-edge (node → ancestor) found, every node on the DFS-tree path from ancestor to node is added to loop_nodes. Correctness note: we track depth to find ancestors unambiguously and use a per-component parent table reset on each new component start. """ all_ids = {n["system_id"] for n in nodes} visited = set() loop_nodes = set() for start_id in all_ids: if start_id in visited: continue # Per-component DFS state parent: dict[str, Optional[str]] = {start_id: None} depth: dict[str, int] = {start_id: 0} # Stack entries: (node_id, parent_id, neighbor_iterator) stack = [(start_id, None, iter(adj.get(start_id, [])))] visited.add(start_id) while stack: node, par, neighbors = stack[-1] try: neighbor = next(neighbors) if neighbor not in visited: visited.add(neighbor) parent[neighbor] = node depth[neighbor] = depth[node] + 1 stack.append((neighbor, node, iter(adj.get(neighbor, [])))) elif neighbor != par and depth.get(neighbor, -1) < depth.get(node, 0): # Back edge to an actual ancestor (not just the tree-parent) # Mark every node on the path from ancestor → node loop_nodes.add(node) loop_nodes.add(neighbor) curr = node while curr != neighbor and curr is not None: loop_nodes.add(curr) curr = parent.get(curr) except StopIteration: stack.pop() return loop_nodes # ── Phase 5: Aperture assignment ─────────────────────────────────────────────── def phase5_apertures( nodes: list[dict], adj: dict, seed_cfg: dict, rng: random.Random, ) -> None: """ Assign aperture_count and gate_connections for each node. All 300 nodes in this graph have horizon stations. """ unused_prob = seed_cfg["augmentation_weights"]["unused_aperture_probability"] for node in nodes: nid = node["system_id"] d = degree(adj, nid) node["gate_connections"] = d # Gateway: 5 apertures (4 active + 1 dormant Sol-facing) if node.get("_gateway"): node["aperture_count"] = 5 node["gate_connections"] = 4 # Sol aperture not traversable continue # Base: apertures = connections (minimum) apertures = d # Narrative texture: some systems have unused apertures # (research interest, mystery, historical significance) if d > 0 and rng.random() < unused_prob: apertures += rng.randint(1, 2) # Cap at 8 (setting limit), but never below actual connections # If degree somehow exceeds 8 (shouldn't happen with tuned augmentation), # we cap gate_connections at 8 as well to maintain consistency. if d > 8: node["gate_connections"] = 8 apertures = min(apertures, 8) # Guarantee: aperture_count >= gate_connections always apertures = max(apertures, node["gate_connections"]) # Floor at 1 (all nodes in this map have stations) apertures = max(apertures, 1) node["aperture_count"] = apertures # ── Phase 6: Validation ──────────────────────────────────────────────────────── def phase6_validate( nodes: list[dict], edges: list[list], adj: dict, seed_cfg: dict, ) -> dict: """ Run all validation checks. Returns a dict of results for the summary report. """ results = {} total = len(nodes) node_map = {n["system_id"]: n for n in nodes} tol = seed_cfg["validation_tolerances"] topology_targets = seed_cfg["topology_targets"] # 1. Connectivity: all nodes reachable from Gateway all_ids = set(n["system_id"] for n in nodes) reachable = bfs_connected(adj, "S-001", all_ids) isolated = all_ids - reachable results["connectivity_ok"] = len(isolated) == 0 results["isolated_count"] = len(isolated) results["isolated_ids"] = sorted(isolated)[:10] # show first 10 if any # 2. Aperture consistency: no system has gate_connections > aperture_count # (Gateway is excluded — it has 4 connections, 5 apertures including Sol) inconsistent = [ n["system_id"] for n in nodes if n["gate_connections"] > n["aperture_count"] ] results["aperture_consistency_ok"] = len(inconsistent) == 0 results["aperture_inconsistent_ids"] = inconsistent[:10] # 3. Topology distribution topology_counts = defaultdict(int) for n in nodes: topology_counts[n["gate_topology"]] += 1 topology_pcts = {k: v / total for k, v in topology_counts.items()} topology_ok = True topology_diffs = {} for topo, target in topology_targets.items(): actual = topology_pcts.get(topo, 0.0) diff = abs(actual - target) topology_diffs[topo] = { "target": target, "actual": round(actual, 3), "count": topology_counts.get(topo, 0), "ok": diff <= tol["topology_target_tolerance_pct"], } if diff > tol["topology_target_tolerance_pct"]: topology_ok = False results["topology_distribution"] = topology_diffs results["topology_ok"] = topology_ok # 4. Hub distribution: at least one hub per sector hubs_per_sector = defaultdict(int) for n in nodes: if n["gate_topology"] == "hub": hubs_per_sector[n["geographic_sector"]] += 1 hub_coverage_ok = all(hubs_per_sector.get(s, 0) >= 1 for s in SECTORS) results["hub_per_sector"] = dict(hubs_per_sector) results["hub_coverage_ok"] = hub_coverage_ok results["hub_total"] = topology_counts.get("hub", 0) results["hub_pct"] = topology_pcts.get("hub", 0.0) results["hub_pct_ok"] = topology_pcts.get("hub", 0.0) <= tol["max_hubs_pct"] # 5. Dead-end + spur coverage dead_spur_pct = (topology_counts.get("dead_end", 0) + topology_counts.get("spur_end", 0)) / total results["dead_spur_pct"] = round(dead_spur_pct, 3) results["dead_spur_ok"] = tol["dead_end_plus_spur_min_pct"] <= dead_spur_pct <= tol["dead_end_plus_spur_max_pct"] # 6. Gateway placement gateway = node_map.get("S-001") gw_connections = degree(adj, "S-001") results["gateway_sector"] = gateway["geographic_sector"] if gateway else "MISSING" results["gateway_topology"] = gateway["gate_topology"] if gateway else "MISSING" results["gateway_apertures"] = gateway["aperture_count"] if gateway else 0 results["gateway_connections_in_adj"] = gw_connections results["gateway_ok"] = ( gateway is not None and gateway["geographic_sector"] == "core" and gw_connections >= tol["gateway_min_connections"] and gw_connections <= tol["gateway_max_connections"] + 1 ) # 7. earth_proximity distribution distances = bfs_distances(adj, "S-001") proximity_counts = defaultdict(int) for nid in all_ids: d = distances.get(nid, 9999) if d <= 2: proximity_counts["immediate"] += 1 elif d <= 10: proximity_counts["proximate"] += 1 elif d <= 30: proximity_counts["distant"] += 1 else: proximity_counts["irrelevant"] += 1 results["earth_proximity_distribution"] = dict(proximity_counts) results["immediate_ok"] = proximity_counts["immediate"] <= tol["earth_proximity_immediate_max"] # Attach hop distance to each node for node in nodes: nid = node["system_id"] d = distances.get(nid, 9999) if d <= 2: node["_earth_proximity"] = "immediate" elif d <= 10: node["_earth_proximity"] = "proximate" elif d <= 30: node["_earth_proximity"] = "distant" else: node["_earth_proximity"] = "irrelevant" node["_hop_distance_from_gateway"] = d # Cross-sector connection counts cross_counts = defaultdict(int) for a_id, b_id in edges: sec_a = node_map[a_id]["geographic_sector"] sec_b = node_map[b_id]["geographic_sector"] if sec_a != sec_b: pair = tuple(sorted([sec_a, sec_b])) cross_counts[pair] += 1 results["cross_sector_connections"] = {f"{a}|{b}": c for (a, b), c in sorted(cross_counts.items())} # Overall pass/fail results["overall_ok"] = all([ results["connectivity_ok"], results["aperture_consistency_ok"], results["hub_coverage_ok"], ]) return results # ── JSON output ──────────────────────────────────────────────────────────────── def build_output_json(nodes: list[dict], edges: list[list]) -> dict: """ Build the canonical star-map.json structure. Strips internal _gateway and _hop_distance fields from output. """ output_nodes = [] for n in nodes: out = { "system_id": n["system_id"], "system_name": n["system_name"], "star_type": n["star_type"], "geographic_sector": n["geographic_sector"], "geographic_band": n["geographic_band"], "political_zone": n["political_zone"], "settlement_wave": n["settlement_wave"], "gate_topology": n["gate_topology"], "aperture_count": n["aperture_count"], "gate_connections": n["gate_connections"], "earth_proximity": n.get("_earth_proximity", "irrelevant"), "hop_distance_from_gateway": n.get("_hop_distance_from_gateway", 9999), } # Mark the Gateway if n.get("_gateway"): out["is_gateway"] = True output_nodes.append(out) return { "_meta": { "generated": "2026-03-13", "version": "0.1", "system_count": len(output_nodes), "edge_count": len(edges), "note": "Placeholder IDs and names. Naming pass required before CSV population.", }, "nodes": output_nodes, "edges": [[a, b] for a, b in edges], } # ── D2 generation ────────────────────────────────────────────────────────────── def d2_safe_id(system_id: str) -> str: """Convert S-001 to s001 for d2 node IDs (no hyphens).""" return system_id.replace("-", "").lower() def generate_sector_d2( sector: str, nodes: list[dict], edges: list[list], node_map: dict, adj: dict, ) -> str: """ Generate d2 source for one sector map. Includes all systems in the sector as full nodes. Cross-sector connections shown as stub nodes. """ sector_ids = {n["system_id"] for n in nodes if n["geographic_sector"] == sector} sector_label = SECTOR_LABELS[sector] # Gather cross-sector stubs needed stub_ids = set() for a_id, b_id in edges: a_sec = node_map[a_id]["geographic_sector"] b_sec = node_map[b_id]["geographic_sector"] if a_id in sector_ids and b_id not in sector_ids: stub_ids.add(b_id) elif b_id in sector_ids and a_id not in sector_ids: stub_ids.add(a_id) lines = [] lines.append(f"# Star Map — {sector_label}") lines.append(f"# Sector map. Cross-sector connections shown as stub nodes (dashed border).") lines.append(f"# Node color = settlement wave. Shape = topology type.") lines.append("") lines.append("vars: {") lines.append(f' bg: "{D2_BG}"') lines.append(f' txt: "{D2_TXT}"') lines.append(f' acc: "{D2_ACC}"') lines.append("}") lines.append("") # Root style lines.append(f"direction: right") lines.append("") lines.append(f'style.fill: "{D2_BG}"') lines.append(f'style.stroke: "{D2_ACC}"') lines.append(f'style.font-color: "{D2_TXT}"') lines.append("") # Legend lines.append("legend: Legend {") lines.append(f' style.fill: "{D2_BG}"; style.stroke: "{D2_ACC}"; style.font-color: "{D2_TXT}"') lines.append(f' style.font-size: 10') for wave, stroke in WAVE_STROKE.items(): fill = WAVE_FILL[wave] w_label = wave.replace("_", " ").title() w_id = wave.replace("_", "") lines.append(f' {w_id}: {w_label} {{ style.fill: "{fill}"; style.stroke: "{stroke}"; style.font-color: "{D2_TXT}" }}') lines.append("}") lines.append("") # Sector nodes for n in sorted(nodes, key=lambda x: x["system_id"]): if n["geographic_sector"] != sector: continue nid = n["system_id"] d2id = d2_safe_id(nid) wave = n["settlement_wave"] topo = n["gate_topology"] fill = WAVE_FILL[wave] stroke = WAVE_STROKE[wave] # Gateway gets special treatment if n.get("_gateway") or nid == "S-001": fill = GATEWAY_FILL stroke = GATEWAY_STROKE label = f"{nid}\\n[GATEWAY]\\n{topo}" else: label = f"{nid}\\n{wave.replace('_', ' ')}\\n{topo}" shape = d2_node_shape(topo) node_line = f'{d2id}: "{label}" {{' lines.append(node_line) lines.append(f' shape: {shape}') lines.append(f' style.fill: "{fill}"') lines.append(f' style.stroke: "{stroke}"') lines.append(f' style.font-color: "{D2_TXT}"') lines.append(f' style.font-size: 9') if nid == "S-001": lines.append(f' style.stroke-width: 3') lines.append("}") lines.append("") # Stub nodes for cross-sector systems for stub_id in sorted(stub_ids): stub_node = node_map[stub_id] d2id = d2_safe_id(stub_id) stub_sector_label = SECTOR_LABELS[stub_node["geographic_sector"]] label = f"{stub_id}\\n[{stub_sector_label}]" lines.append(f'{d2id}: "{label}" {{') lines.append(f' shape: rectangle') lines.append(f' style.fill: "{STUB_FILL}"') lines.append(f' style.stroke: "{STUB_STROKE}"') lines.append(f' style.stroke-dash: 5') lines.append(f' style.font-color: "{STUB_STROKE}"') lines.append(f' style.font-size: 9') lines.append("}") lines.append("") # Edges rendered_edges = set() for a_id, b_id in edges: a_sec = node_map[a_id]["geographic_sector"] b_sec = node_map[b_id]["geographic_sector"] # Only render edges where at least one endpoint is in this sector if a_id not in sector_ids and b_id not in sector_ids: continue edge_key = tuple(sorted([a_id, b_id])) if edge_key in rendered_edges: continue rendered_edges.add(edge_key) d2a = d2_safe_id(a_id) d2b = d2_safe_id(b_id) is_cross = a_sec != b_sec is_gateway_edge = (a_id == "S-001" or b_id == "S-001") if is_gateway_edge: color = EDGE_COLOR_GATEWAY elif is_cross: color = EDGE_COLOR_CROSS else: color = EDGE_COLOR_INTRA edge_line = f"{d2a} -- {d2b}" if is_cross: lines.append(f"{edge_line}: {{") lines.append(f' style.stroke: "{color}"') lines.append(f' style.stroke-dash: 4') lines.append(f' style.stroke-width: 1') lines.append("}") else: lines.append(f"{edge_line}: {{") lines.append(f' style.stroke: "{color}"') lines.append("}") lines.append("") return "\n".join(lines) def generate_overview_d2( nodes: list[dict], edges: list[list], node_map: dict, ) -> str: """ Generate the overview d2 showing sectors as cluster nodes with inter-sector edge counts. """ # Count cross-sector connections cross_counts = defaultdict(int) sector_node_counts = defaultdict(int) for n in nodes: sector_node_counts[n["geographic_sector"]] += 1 for a_id, b_id in edges: sec_a = node_map[a_id]["geographic_sector"] sec_b = node_map[b_id]["geographic_sector"] if sec_a != sec_b: pair = tuple(sorted([sec_a, sec_b])) cross_counts[pair] += 1 # Hub counts per sector hub_counts = defaultdict(int) for n in nodes: if n["gate_topology"] == "hub": hub_counts[n["geographic_sector"]] += 1 lines = [] lines.append("# Star Map — Overview") lines.append("# Sector cluster view. Nodes = sectors. Edge labels = cross-sector gate connections.") lines.append("") lines.append("vars: {") lines.append(f' bg: "{D2_BG}"') lines.append(f' txt: "{D2_TXT}"') lines.append(f' acc: "{D2_ACC}"') lines.append("}") lines.append("") lines.append(f'direction: right') lines.append(f'style.fill: "{D2_BG}"') lines.append(f'style.stroke: "{D2_ACC}"') lines.append(f'style.font-color: "{D2_TXT}"') lines.append("") # Sector nodes sector_colors = { "core": ("#1a2040", "#3060c0"), "north_reach": ("#1a2820", "#4a9060"), "west_reach": ("#201e14", "#907030"), "south_reach": ("#201814", "#905030"), "east_reach": ("#1a2028", "#4070a0"), "deep_frontier": ("#201010", "#803030"), } for sector in SECTORS: label = SECTOR_LABELS[sector] count = sector_node_counts[sector] hubs = hub_counts[sector] fill, stroke = sector_colors[sector] d2id = sector.replace("_", "") lines.append(f'{d2id}: "{label}\\n{count} systems · {hubs} hubs" {{') lines.append(f' shape: rectangle') lines.append(f' style.fill: "{fill}"') lines.append(f' style.stroke: "{stroke}"') lines.append(f' style.font-color: "{D2_TXT}"') lines.append(f' style.border-radius: 8') lines.append("}") lines.append("") # Special Gateway callout lines.append('gateway_note: "GATEWAY (S-001)\\nDiplomatic Periphery · Core\\nSol aperture: dormant" {') lines.append(f' shape: hexagon') lines.append(f' style.fill: "{GATEWAY_FILL}"') lines.append(f' style.stroke: "{GATEWAY_STROKE}"') lines.append(f' style.font-color: "{D2_TXT}"') lines.append(f' style.stroke-width: 3') lines.append("}") lines.append(f'gateway_note -> core: "located in" {{') lines.append(f' style.stroke: "{GATEWAY_STROKE}"; style.stroke-dash: 3') lines.append("}") lines.append("") # Cross-sector edges rendered = set() for (sec_a, sec_b), count in sorted(cross_counts.items()): pair_key = (sec_a, sec_b) if pair_key in rendered: continue rendered.add(pair_key) d2a = sec_a.replace("_", "") d2b = sec_b.replace("_", "") lines.append(f'{d2a} -- {d2b}: "{count} connections" {{') lines.append(f' style.stroke: "{EDGE_COLOR_CROSS}"') lines.append(f' style.font-color: "{D2_TXT}"') lines.append("}") lines.append("") return "\n".join(lines) # ── Validation report printer ────────────────────────────────────────────────── def print_validation_summary(nodes: list[dict], edges: list[list], validation: dict) -> None: print("\n" + "=" * 60) print("STAR MAP GENERATION — VALIDATION SUMMARY") print("=" * 60) print(f" Systems: {len(nodes)}") print(f" Edges: {len(edges)}") print() # Connectivity ok = "OK" if validation["connectivity_ok"] else "FAIL" print(f" Connectivity: [{ok}] all nodes reachable from Gateway") if not validation["connectivity_ok"]: print(f" Isolated: {validation['isolated_count']} nodes: {validation['isolated_ids']}") # Aperture consistency ok = "OK" if validation["aperture_consistency_ok"] else "FAIL" print(f" Aperture consistency: [{ok}]") if not validation["aperture_consistency_ok"]: print(f" Inconsistent: {validation['aperture_inconsistent_ids']}") # Topology distribution print() print(" Topology distribution:") topo_ok_all = True for topo, info in sorted(validation["topology_distribution"].items()): flag = "ok" if info["ok"] else "WARN" print(f" {topo:<16} target={info['target']:.0%} actual={info['actual']:.1%} ({info['count']:3d} systems) [{flag}]") if not info["ok"]: topo_ok_all = False if not topo_ok_all: print(" Note: topology targets are soft. Deviation within 4pp is expected.") # Hub distribution print() print(" Hub distribution per sector:") for sector in SECTORS: count = validation["hub_per_sector"].get(sector, 0) flag = "ok" if count >= 1 else "WARN" print(f" {sector:<20} {count} hubs [{flag}]") print(f" Total hubs: {validation['hub_total']} ({validation['hub_pct']:.1%})") # Dead-end / spur ok = "ok" if validation["dead_spur_ok"] else "WARN" print() print(f" Dead-end + spur coverage: {validation['dead_spur_pct']:.1%} [{ok}]") # Gateway print() ok = "OK" if validation["gateway_ok"] else "FAIL" print(f" Gateway (S-001): [{ok}]") print(f" Sector: {validation['gateway_sector']}") print(f" Topology: {validation['gateway_topology']}") print(f" Apertures: {validation['gateway_apertures']}") print(f" Adj degree: {validation['gateway_connections_in_adj']}") # Earth proximity print() print(" Earth proximity distribution (hop distance from Gateway):") for zone, count in sorted(validation["earth_proximity_distribution"].items()): print(f" {zone:<12} {count:3d} systems") # Cross-sector connections print() print(" Cross-sector connections:") for pair, count in sorted(validation["cross_sector_connections"].items()): print(f" {pair:<40} {count}") # Overall print() overall = "PASS" if validation["overall_ok"] else "ISSUES DETECTED" print(f" Overall: {overall}") print("=" * 60) print() # ── Main ─────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Generate Settled Reach star map") parser.add_argument( "--seed-file", default=str(DEFAULT_SEED_FILE), help=f"Path to seed configuration JSON (default: {DEFAULT_SEED_FILE})", ) args = parser.parse_args() seed_file = Path(args.seed_file) if not seed_file.exists(): print(f"ERROR: seed file not found: {seed_file}", file=sys.stderr) sys.exit(1) with open(seed_file, "r") as f: seed_cfg = json.load(f) # Strip _comment keys recursively so they don't pollute dict iterations def strip_comments(obj): if isinstance(obj, dict): return {k: strip_comments(v) for k, v in obj.items() if not k.startswith("_comment")} if isinstance(obj, list): return [strip_comments(item) for item in obj] return obj seed_cfg = strip_comments(seed_cfg) # Initialize RNG with fixed seed for determinism rng = random.Random(seed_cfg["random_seed"]) print(f"Loaded seed config: {seed_file}") print(f"Random seed: {seed_cfg['random_seed']}") print(f"Target system count: {seed_cfg['system_count']}") # Phase 1: Place systems print("Phase 1: Placing systems...") nodes = phase1_place_systems(seed_cfg, rng) print(f" Placed {len(nodes)} systems") # Phase 2: Spanning tree print("Phase 2: Building spanning tree backbone...") edges, edges_set, adj = phase2_spanning_tree(nodes, seed_cfg, rng) print(f" Spanning tree: {len(edges)} edges") # Phase 3: Augmentation print("Phase 3: Augmentation passes (hubs, loops, bridges)...") phase3_augment(nodes, edges, edges_set, adj, seed_cfg, rng) print(f" Post-augmentation: {len(edges)} edges") # Phase 4: Classify topology print("Phase 4: Classifying topology...") phase4_classify_topology(nodes, edges, adj) # Phase 5: Aperture assignment print("Phase 5: Assigning apertures...") phase5_apertures(nodes, adj, seed_cfg, rng) # Phase 6: Validation print("Phase 6: Validating...") node_map = {n["system_id"]: n for n in nodes} validation = phase6_validate(nodes, edges, adj, seed_cfg) # Print validation summary print_validation_summary(nodes, edges, validation) # Write star-map.json OUTPUT_JSON.parent.mkdir(parents=True, exist_ok=True) output_data = build_output_json(nodes, edges) with open(OUTPUT_JSON, "w") as f: json.dump(output_data, f, indent=2) print(f"Written: {OUTPUT_JSON}") # Write sector d2 files OUTPUT_D2_DIR.mkdir(parents=True, exist_ok=True) sector_filenames = { "core": "star-map-core.d2", "north_reach": "star-map-north-reach.d2", "west_reach": "star-map-west-reach.d2", "south_reach": "star-map-south-reach.d2", "east_reach": "star-map-east-reach.d2", "deep_frontier": "star-map-deep-frontier.d2", } for sector, filename in sector_filenames.items(): d2_content = generate_sector_d2(sector, nodes, edges, node_map, adj) out_path = OUTPUT_D2_DIR / filename with open(out_path, "w") as f: f.write(d2_content) sector_count = sum(1 for n in nodes if n["geographic_sector"] == sector) print(f"Written: {out_path} ({sector_count} systems)") # Write overview d2 overview_content = generate_overview_d2(nodes, edges, node_map) overview_path = OUTPUT_D2_DIR / "star-map-overview.d2" with open(overview_path, "w") as f: f.write(overview_content) print(f"Written: {overview_path}") # Summary if validation["overall_ok"]: print("\nGeneration complete. All critical checks passed.") else: print("\nGeneration complete with issues. Review validation output above.") if validation["isolated_count"] > 0: print(f" ACTION NEEDED: {validation['isolated_count']} isolated systems must be connected.") return 0 if validation["overall_ok"] else 1 if __name__ == "__main__": sys.exit(main())