settled-reach/tooling/generate-star-map-data.py

#!/usr/bin/env python3
"""Generate client/data/star_map_data.json from star-map.json + systems.db + wiki.

Run from any directory — paths are resolved relative to this script's location:
    python3 tooling/generate-star-map-data.py
    python3 tooling/generate-star-map-data.py --check   # exit 1 if committed JSON is stale

Sources:
    docs/design/star-map.json     — graph topology (nodes + edges)
    server/data/systems.db        — proper names, geographic sectors, bodies, GDP tier
    wiki/star-systems/            — star type, GTTR excerpt (bodies/population from systems.db)

Output:
    client/data/star_map_data.json — self-contained client data for the star map UI
"""

import json
import os
import re
import sqlite3
import sys
import tempfile

# Resolve project root from this script's location.
# Works regardless of cwd — no fragile relative path guessing.
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_PROJECT_ROOT = os.path.dirname(_SCRIPT_DIR)

STAR_MAP_PATH = os.path.join(_PROJECT_ROOT, "docs", "design", "star-map.json")
SYSTEMS_DB_PATH = os.path.join(_PROJECT_ROOT, "server", "data", "systems.db")
WIKI_PATH = os.path.join(_PROJECT_ROOT, "wiki", "star-systems")
OUTPUT_PATH = os.path.join(_PROJECT_ROOT, "client", "data", "star_map_data.json")


def system_id_to_wiki_slug(system_id: str) -> str:
    """Convert system_id to wiki folder slug. 'GJ 71' → 'GJ-71'."""
    return system_id.replace(" ", "-")


def parse_wiki_index(system_id: str) -> dict:
    """Extract star type from index.md.

    Bodies and population are authoritative from systems.db — not read from wiki.
    Returns dict with key: star_type (string, may be empty).
    """
    slug = system_id_to_wiki_slug(system_id)
    path = os.path.join(WIKI_PATH, slug, "index.md")
    result = {"star_type": ""}
    if not os.path.exists(path):
        return result
    with open(path, encoding="utf-8") as f:
        content = f.read()

    # Star row: | **Star** | G8V · 11.9 ly |
    m = re.search(r"\|\s*\*\*Star\*\*\s*\|\s*([^|]+?)\s*\|", content)
    if m:
        raw = m.group(1).strip()
        # Extract spectral class — everything before " ·" or end of string
        result["star_type"] = raw.split("·")[0].strip()

    return result


def parse_gttr_excerpt(system_id: str) -> str:
    """Extract the first body paragraph from gttr.md as the GTTR excerpt."""
    slug = system_id_to_wiki_slug(system_id)
    path = os.path.join(WIKI_PATH, slug, "gttr.md")
    if not os.path.exists(path):
        return ""
    with open(path, encoding="utf-8") as f:
        lines = f.readlines()

    # Skip the H1 heading line, then find the first non-empty paragraph
    in_content = False
    paragraph_lines = []
    for line in lines:
        stripped = line.strip()
        if not in_content:
            # Skip heading and blank lines at start
            if stripped.startswith("# "):
                in_content = True
            continue
        if not stripped:
            # Blank line ends the paragraph if we've collected lines
            if paragraph_lines:
                break
        else:
            if not stripped.startswith("#"):
                paragraph_lines.append(stripped)

    return " ".join(paragraph_lines)


GDP_PER_CAPITA: dict = {
    5: 75_000,
    4: 40_000,
    3: 15_000,
    2: 5_000,
    1: 2_000,
    0: 500,
}


def infer_tier_from_population(pop: int) -> int:
    """Infer an economic tier from total population when no explicit tier is set."""
    if pop >= 5_000_000_000:
        return 5
    if pop >= 1_000_000_000:
        return 4
    if pop >= 200_000_000:
        return 3
    if pop >= 10_000_000:
        return 2
    return 1


def compute_gdp(total_pop: int, economic_tier: int | None) -> str:
    """Return a formatted GDP string in Tractus, or empty string if no population."""
    if total_pop == 0:
        return ""
    tier = economic_tier if economic_tier is not None else infer_tier_from_population(total_pop)
    per_cap = GDP_PER_CAPITA.get(tier, GDP_PER_CAPITA[1])
    value = total_pop * per_cap
    if value < 1_000_000_000:
        return f"{value / 1_000_000:.1f} MTr"
    if value < 1_000_000_000_000:
        return f"{value / 1_000_000_000:.1f} BTr"
    if value < 1_000_000_000_000_000:
        return f"{value / 1_000_000_000_000:.1f} TTr"
    return f"{value / 1_000_000_000_000_000:.1f} QTr"


def build_adjacency(edges: list) -> dict:
    """Build a map from system_id to list of adjacent system_ids from edges."""
    adj: dict = {}
    for edge in edges:
        if len(edge) < 2:
            continue
        a, b = edge[0], edge[1]
        adj.setdefault(a, [])
        adj.setdefault(b, [])
        if b not in adj[a]:
            adj[a].append(b)
        if a not in adj[b]:
            adj[b].append(a)
    return adj


def generate() -> dict:
    """Generate the enriched star map data dict."""
    if not os.path.exists(STAR_MAP_PATH):
        print(f"ERROR: star-map.json not found at {STAR_MAP_PATH}", file=sys.stderr)
        sys.exit(1)
    if not os.path.exists(SYSTEMS_DB_PATH):
        print(f"ERROR: systems.db not found at {SYSTEMS_DB_PATH}", file=sys.stderr)
        sys.exit(1)

    with open(STAR_MAP_PATH) as f:
        star_map = json.load(f)

    try:
        conn = sqlite3.connect(SYSTEMS_DB_PATH)
        conn.row_factory = sqlite3.Row
        cur = conn.cursor()

        cur.execute(
            "SELECT system_id, proper_name, geographic_sector, geographic_band, currency_zone "
            "FROM star_systems"
        )
        db_lookup = {row["system_id"]: dict(row) for row in cur.fetchall()}

        # Aggregate body data per system from the bodies table
        cur.execute("""
            SELECT system_id,
                SUM(CASE WHEN atmosphere IN ('breathable','standard') AND body_type IN ('planet','moon') THEN 1 ELSE 0 END) AS habitable,
                SUM(CASE WHEN inhabited = 1 THEN 1 ELSE 0 END) AS inhabited,
                SUM(CASE WHEN inhabited = 1 THEN COALESCE(population, 0) ELSE 0 END) AS total_pop
            FROM bodies
            GROUP BY system_id
        """)
        body_stats = {row["system_id"]: dict(row) for row in cur.fetchall()}

        # Also sum station populations
        cur.execute("""
            SELECT system_id,
                SUM(COALESCE(population, 0)) AS station_pop
            FROM stations
            GROUP BY system_id
        """)
        station_stats = {row["system_id"]: dict(row) for row in cur.fetchall()}

        # Economic tier for GDP calculation
        cur.execute("SELECT system_id, economic_tier FROM system_economy")
        econ_tiers = {row["system_id"]: row["economic_tier"] for row in cur.fetchall()}

        # Per-system body list for atlas orbital diagram (D-191 §6)
        cur.execute("""
            SELECT system_id, body_id, parent_body_id, orbit_index, body_type,
                   proper_name, atmosphere, inhabited, population,
                   terrain_reference, mass_class, body_radius_km
            FROM bodies
            ORDER BY system_id, orbit_index
        """)
        orbit_bodies_by_system: dict = {}
        for row in cur.fetchall():
            sid_ = row["system_id"]
            orbit_bodies_by_system.setdefault(sid_, []).append({
                "body_id": row["body_id"],
                "parent_body_id": row["parent_body_id"],
                "orbit_index": row["orbit_index"],
                "body_type": row["body_type"],
                "proper_name": row["proper_name"],
                "atmosphere": row["atmosphere"],
                "inhabited": bool(row["inhabited"]),
                "population": int(row["population"] or 0),
                "terrain_reference": row["terrain_reference"],
                "mass_class": row["mass_class"],
                "body_radius_km": float(row["body_radius_km"] or 0),
            })

        # Per-system station list for atlas orbital diagram (D-191 §6)
        cur.execute("""
            SELECT system_id, station_id, orbits_body_id, station_type,
                   proper_name, population, governance_type, economic_role
            FROM stations
            ORDER BY system_id
        """)
        stations_by_system: dict = {}
        for row in cur.fetchall():
            sid_ = row["system_id"]
            stations_by_system.setdefault(sid_, []).append({
                "station_id": row["station_id"],
                "orbits_body_id": row["orbits_body_id"],
                "station_type": row["station_type"],
                "proper_name": row["proper_name"],
                "population": int(row["population"] or 0),
                "governance_type": row["governance_type"],
                "economic_role": row["economic_role"],
            })
    except sqlite3.Error as e:
        print(f"ERROR: systems.db query failed: {e}", file=sys.stderr)
        sys.exit(1)
    finally:
        conn.close()

    adjacency = build_adjacency(star_map["edges"])

    nodes = []
    wiki_hits = 0
    gttr_hits = 0
    for n in star_map["nodes"]:
        sid = n["system_id"]
        db = db_lookup.get(sid, {})
        wiki = parse_wiki_index(sid)
        gttr = parse_gttr_excerpt(sid)
        if wiki["star_type"]:
            wiki_hits += 1
        if gttr:
            gttr_hits += 1

        entry = {
            "system_id": sid,
            "proper_name": db.get("proper_name", ""),
            "geographic_sector": db.get("geographic_sector", "unknown"),
            "geographic_band": db.get("geographic_band", ""),
            "gate_topology": n["gate_topology"],
            "aperture_count": n["aperture_count"],
            "gate_connections": n["gate_connections"],
            "hop_distance": n["hop_distance_from_gateway"],
            "adjacent_systems": adjacency.get(sid, []),
        }
        if wiki["star_type"]:
            entry["star_type"] = wiki["star_type"]

        # Currency zone from star_systems
        cz = db.get("currency_zone", "")
        if cz:
            entry["currency_zone"] = cz

        # Bodies + population from systems.db (authoritative, not wiki)
        bs = body_stats.get(sid, {})
        ss = station_stats.get(sid, {})
        hab = int(bs.get("habitable", 0))
        inh = int(bs.get("inhabited", 0))
        body_pop = int(bs.get("total_pop", 0))
        sta_pop = int(ss.get("station_pop", 0))
        total_pop = body_pop + sta_pop

        entry["bodies"] = "%d habitable · %d inhabited" % (hab, inh)
        entry["population"] = "{:,}".format(total_pop)
        gdp_str = compute_gdp(total_pop, econ_tiers.get(sid))
        if gdp_str:
            entry["gdp"] = gdp_str
        if gttr:
            entry["gttr_excerpt"] = gttr
        if n.get("is_gateway"):
            entry["is_gateway"] = True

        # Atlas orbital diagram data (D-191 §6) — per-body and per-station arrays
        entry["orbit_bodies"] = orbit_bodies_by_system.get(sid, [])
        entry["stations"] = stations_by_system.get(sid, [])

        nodes.append(entry)

    nodes.sort(key=lambda x: (x["hop_distance"], x["system_id"]))

    print(f"  Wiki index data: {wiki_hits}/{len(nodes)} systems")
    print(f"  GTTR excerpts:   {gttr_hits}/{len(nodes)} systems")

    return {
        "_meta": {
            "generated_from": "star-map.json + systems.db + wiki/star-systems",
            "system_count": len(nodes),
            "edge_count": len(star_map["edges"]),
            "note": "Client-side star map + atlas orbital data. Regenerate with: tooling/generate-star-map-data.py",
        },
        "nodes": nodes,
        "edges": star_map["edges"],
    }


def write_output(data: dict, path: str) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
        f.write("\n")


def main() -> None:
    check_mode = "--check" in sys.argv

    data = generate()

    if check_mode:
        # Generate to temp file and compare against committed JSON
        if not os.path.exists(OUTPUT_PATH):
            print(f"STALE: {OUTPUT_PATH} does not exist — run without --check to generate")
            sys.exit(1)
        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp:
            json.dump(data, tmp, indent=2, ensure_ascii=False)
            tmp.write("\n")
            tmp_path = tmp.name
        try:
            with open(tmp_path) as a, open(OUTPUT_PATH) as b:
                if a.read() != b.read():
                    print(f"STALE: {OUTPUT_PATH} differs from generated output")
                    print("Run: python3 tooling/generate-star-map-data.py")
                    sys.exit(1)
            print(f"OK: {OUTPUT_PATH} is up to date")
        finally:
            os.unlink(tmp_path)
    else:
        write_output(data, OUTPUT_PATH)
        print(f"Generated {OUTPUT_PATH}")
        print(f"  Nodes: {data['_meta']['system_count']}, Edges: {data['_meta']['edge_count']}")


if __name__ == "__main__":
    main()