jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(simulation): extend economy-db pipeline with corporation sync and validation (#805)

Browse Source 
Extends import_economics.py from 6-step to 8-step pipeline:
- Loads wiki corporation markdown frontmatter as authoritative source
- Syncs corporations table (hard error on proper_name divergence per D-182)
- Populates corp_presence table (one row per corp × headquarters system)
- Splits validation: structural checks block commit; coverage checks post-commit
- D-175 Phase 2 gate: 3+ corps per commodity, 1+ corp per system with pop > 100K

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-04-07 13:55:40 +02:00
co-authored by Claude Sonnet 4.6
parent 4d1b0efbbe
commit 3447b38bcb
2 changed files with 355 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/data/systems.db
BIN
View File
Binary file not shown.
tooling/economy-db/import_economics.py
+355 -19
View File
@@ -3,12 +3,19 @@
Import economics data into systems.db.
Reads TOML/JSON source files and populates the economics tables:
- gate_links from docs/design/star-map.json (335 edges, bidirectional)
- commodities from wiki/economics/commodities.toml (36 types)
- gate_links from docs/design/star-map.json (335 edges, bidirectional)
- commodities from wiki/economics/commodities.toml (36 types)
- production_chains + chain_inputs from wiki/economics/production_chains.toml
- currency_zone on star_systems (default TRACTUS_PRIMARY)
- currency_zone on star_systems (default TRACTUS_PRIMARY)
- gate_energy_connected on star_systems (D-186: false for MARK_PRIMARY zones)
- corporations from wiki/corporations/*.md (sync + insert new records)
- corp_presence from wiki/corporations/*.md (headquarters location data)
Does NOT populate corp_presence — that's a future pipeline step.
Validation (hard errors, non-zero exit on any failure):
- Wiki corporation names must match DB proper_name records (D-182 sync constraint)
- Chain completeness: every intermediate commodity has at least one production chain
- Commodity coverage: 3+ corporations per major commodity type (D-175)
- System coverage: 1+ corporation per inhabited system with population > 100K (D-175)
Usage:
python3 tooling/economy-db/import_economics.py
@@ -18,6 +25,7 @@ Usage:
import argparse
import json
import re
import sqlite3
import sys
import tomllib
@@ -30,6 +38,7 @@ STAR_MAP = REPO_ROOT / "docs" / "design" / "star-map.json"
COMMODITIES_TOML = REPO_ROOT / "wiki" / "economics" / "commodities.toml"
CHAINS_TOML = REPO_ROOT / "wiki" / "economics" / "production_chains.toml"
SCHEMA_SQL = REPO_ROOT / "server" / "data" / "systems-schema.sql"
CORPORATIONS_DIR = REPO_ROOT / "wiki" / "corporations"
# ---------------------------------------------------------------------------
@@ -102,6 +111,7 @@ CREATE INDEX IF NOT EXISTS idx_corp_presence_location ON corp_presence(location_
# Columns to add to existing tables (ALTER TABLE is idempotent via try/except)
COLUMN_MIGRATIONS = [
("star_systems", "currency_zone", "TEXT DEFAULT 'TRACTUS_PRIMARY'"),
("star_systems", "gate_energy_connected", "INTEGER DEFAULT 1"),
("corporations", "behavioral_archetype", "TEXT"),
("corporations", "supply_chain_role", "TEXT"),
("corporations", "shadow_economy_access", "INTEGER DEFAULT 0"),
@@ -263,11 +273,211 @@ def set_currency_zones(conn: sqlite3.Connection, dry_run: bool) -> dict:
return counts
# ---------------------------------------------------------------------------
# Gate energy connectivity (D-186)
# ---------------------------------------------------------------------------
def set_gate_energy(conn: sqlite3.Connection, dry_run: bool) -> dict:
"""Set gate_energy_connected on star_systems based on currency_zone.
MARK_PRIMARY zones default to false (Compact refused Gate Corp dependency).
All other zones default to true.
"""
if dry_run:
return {"on_grid": "non-MARK_PRIMARY", "off_grid": "MARK_PRIMARY"}
# Default: all systems on-grid
conn.execute("UPDATE star_systems SET gate_energy_connected = 1 WHERE gate_energy_connected IS NULL")
# MARK_PRIMARY zones are off-grid (Compact energy sovereignty)
conn.execute("UPDATE star_systems SET gate_energy_connected = 0 WHERE currency_zone = 'MARK_PRIMARY'")
counts = {}
for row in conn.execute(
"SELECT gate_energy_connected, COUNT(*) FROM star_systems GROUP BY gate_energy_connected"
):
label = "on_grid" if row[0] == 1 else "off_grid"
counts[label] = row[1]
return counts
# ---------------------------------------------------------------------------
# Corporation wiki parsing
# ---------------------------------------------------------------------------
def _parse_corp_frontmatter(path: Path) -> dict | None:
"""Parse YAML frontmatter from a wiki corporation markdown file."""
text = path.read_text()
lines = text.split("\n")
if not lines or lines[0].strip() != "---":
return None
end_idx = None
for i, line in enumerate(lines[1:], 1):
if line.strip() == "---":
end_idx = i
break
if end_idx is None:
return None
fm: dict = {}
for line in lines[1:end_idx]:
if ":" not in line:
continue
key, _, val = line.partition(":")
key = key.strip()
val = val.strip()
if val.startswith("[") and val.endswith("]"):
items = [x.strip().strip('"').strip("'") for x in val[1:-1].split(",")]
fm[key] = [item for item in items if item]
else:
fm[key] = val.strip('"').strip("'")
return fm
def load_wiki_corps() -> list[dict]:
"""Load all wiki corporation files. Returns list of parsed corp records."""
corps = []
for md_file in sorted(CORPORATIONS_DIR.glob("*.md")):
if md_file.name == "index.md":
continue
fm = _parse_corp_frontmatter(md_file)
if not fm or not fm.get("slug") or not fm.get("title"):
continue
hq = fm.get("headquarters", "")
m = re.search(r"\(([^)]+)\)", hq)
system_id = m.group(1) if m else None
corps.append({
"corp_id": fm["slug"],
"proper_name": fm["title"],
"system_id": system_id,
"tags": fm.get("tags", []),
"scope": fm.get("scope", ""),
})
return corps
# ---------------------------------------------------------------------------
# Corporation sync (D-182: wiki is source of truth)
# ---------------------------------------------------------------------------
def sync_corporations(
conn: sqlite3.Connection, wiki_corps: list[dict], dry_run: bool
) -> list[str]:
"""Sync wiki corps to DB. Hard error on proper_name divergence (D-182).
Returns list of error strings. Inserts corps that exist in wiki but not DB.
Corps that exist only in DB (legacy records) are left untouched.
headquarters_system is only written if the system_id exists in star_systems
(to avoid FK violations when atlas hasn't yet registered the system).
"""
errors: list[str] = []
existing = {
r[0]: r[1]
for r in conn.execute("SELECT corp_id, proper_name FROM corporations").fetchall()
}
valid_systems = {
r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
}
to_insert = []
for corp in wiki_corps:
corp_id = corp["corp_id"]
proper_name = corp["proper_name"]
if corp_id in existing:
if existing[corp_id] != proper_name:
errors.append(
f"name divergence: corp_id='{corp_id}' "
f"wiki='{proper_name}' db='{existing[corp_id]}'"
)
else:
system_id = corp.get("system_id")
hq_system = system_id if system_id and system_id in valid_systems else None
if system_id and system_id not in valid_systems:
print(f" warning: {corp_id} HQ system '{system_id}' not in DB, "
f"headquarters_system set to NULL")
to_insert.append((
corp_id,
proper_name,
"corporation",
corp.get("scope") or None,
hq_system,
))
if not dry_run and not errors:
conn.executemany(
"""INSERT OR IGNORE INTO corporations
(corp_id, proper_name, corp_type, scope, headquarters_system)
VALUES (?, ?, ?, ?, ?)""",
to_insert,
)
return errors
# ---------------------------------------------------------------------------
# Corp presence population
# ---------------------------------------------------------------------------
def import_corp_presence(
conn: sqlite3.Connection,
wiki_corps: list[dict],
commodity_ids: set[str],
dry_run: bool,
) -> int:
"""Populate corp_presence from wiki headquarters data.
Each corporation gets one presence row at its headquarters system.
primary_operation is set to the first commodity tag that matches a known
commodity ID, or None if no commodity tags are present.
"""
valid_systems = {
r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
}
rows = []
skipped = []
for corp in wiki_corps:
system_id = corp.get("system_id")
if not system_id:
skipped.append(f"{corp['corp_id']} (no headquarters system parsed)")
continue
if system_id not in valid_systems:
skipped.append(f"{corp['corp_id']} (system '{system_id}' not in DB)")
continue
primary_op = next(
(tag for tag in corp.get("tags", []) if tag in commodity_ids), None
)
rows.append((corp["corp_id"], system_id, "system", primary_op))
if skipped:
for s in skipped:
print(f" warning: skipped corp_presence for {s}")
if not dry_run:
conn.execute("DELETE FROM corp_presence")
conn.executemany(
"""INSERT OR IGNORE INTO corp_presence
(corp_id, location_id, location_type, primary_operation)
VALUES (?, ?, ?, ?)""",
rows,
)
return len(rows)
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def validate(conn: sqlite3.Connection) -> list[str]:
"""Validate structural integrity of imported data.
Checks FK integrity, chain commodity references, and chain completeness.
These are hard blockers — broken data must not be committed.
Coverage validation (commodity/system thresholds) is separate and runs
after commit via _validate_commodity_coverage() and _validate_system_coverage().
"""
errors = []
# FK integrity
@@ -297,9 +507,75 @@ def validate(conn: sqlite3.Connection) -> list[str]:
for chain_id, cid in orphan_outputs:
errors.append(f"production_chains: chain '{chain_id}' outputs unknown commodity '{cid}'")
# Chain completeness: every intermediate commodity must have at least one producer
missing_chains = conn.execute("""
SELECT c.commodity_id, c.name
FROM commodities c
WHERE c.tier = 'intermediate'
AND c.commodity_id NOT IN (SELECT output_commodity_id FROM production_chains)
ORDER BY c.commodity_id
""").fetchall()
for cid, name in missing_chains:
errors.append(f"chain completeness: no production chain produces intermediate '{cid}' ({name})")
return errors
def _validate_commodity_coverage(
conn: sqlite3.Connection, wiki_corps: list[dict], commodity_ids: set[str]
) -> list[str]:
"""3+ corporations per major commodity type (raw + intermediate). D-175."""
errors: list[str] = []
major = [
r[0]
for r in conn.execute(
"SELECT commodity_id FROM commodities "
"WHERE tier IN ('raw', 'intermediate') ORDER BY commodity_id"
).fetchall()
]
# Build commodity → corp set from wiki tags filtered to known commodity IDs
coverage: dict[str, set[str]] = {cid: set() for cid in major}
for corp in wiki_corps:
for tag in corp.get("tags", []):
if tag in coverage:
coverage[tag].add(corp["corp_id"])
for cid in major:
n = len(coverage[cid])
if n < 3:
corp_list = sorted(coverage[cid]) if coverage[cid] else ["none"]
errors.append(
f"commodity coverage: '{cid}' has {n}/3 corp(s) — {corp_list}"
)
return errors
def _validate_system_coverage(
conn: sqlite3.Connection, wiki_corps: list[dict]
) -> list[str]:
"""1+ corporation per inhabited system with population > 100K. D-175.
Uses wiki_corps headquarters data (not DB corp_presence) so this check
is accurate in both dry-run and real-run modes.
"""
covered = {c["system_id"] for c in wiki_corps if c.get("system_id")}
populated = conn.execute("""
SELECT se.system_id, ss.proper_name, se.population
FROM system_economy se
JOIN star_systems ss ON se.system_id = ss.system_id
WHERE se.population > 100000
ORDER BY se.system_id
""").fetchall()
return [
f"system coverage: no corp presence in '{sid}' ({name}, pop={pop:,})"
for sid, name, pop in populated
if sid not in covered
]
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
@@ -321,66 +597,126 @@ def main():
print(f" Mode: DRY RUN")
print()
# Load wiki corps before opening DB — allows early exit on parse failures
print(" Loading wiki corporations...")
wiki_corps = load_wiki_corps()
print(f" {len(wiki_corps)} corporation files parsed")
conn = sqlite3.connect(str(db_path))
conn.execute("PRAGMA foreign_keys=ON")
# 1. Migrate schema
print(" [1/5] Schema migration...")
print(" [1/8] Schema migration...")
for table, col, col_type in COLUMN_MIGRATIONS:
_add_column(conn, table, col, col_type)
conn.executescript(MIGRATION_SQL)
print(" tables and columns ready")
# Clear economics tables in FK-safe order (children before parents)
# corp_presence cleared here; corporations table is append-only (never cleared)
if not args.dry_run:
conn.execute("DELETE FROM corp_presence")
conn.execute("DELETE FROM chain_inputs")
conn.execute("DELETE FROM production_chains")
conn.execute("DELETE FROM commodities")
conn.execute("DELETE FROM gate_links")
# 2. Gate links
print(" [2/5] Importing gate links...")
print(" [2/8] Importing gate links...")
n_links = import_gate_links(conn, args.dry_run)
print(f" {n_links} rows (bidirectional)")
# 3. Commodities
print(" [3/5] Importing commodities...")
print(" [3/8] Importing commodities...")
n_commodities = import_commodities(conn, args.dry_run)
print(f" {n_commodities} commodities")
# 4. Production chains
print(" [4/5] Importing production chains...")
print(" [4/8] Importing production chains...")
n_chains, n_inputs = import_chains(conn, args.dry_run)
print(f" {n_chains} chains, {n_inputs} inputs")
# 5. Currency zones
print(" [5/5] Setting currency zones...")
print(" [5/8] Setting currency zones...")
zones = set_currency_zones(conn, args.dry_run)
for zone, count in sorted(zones.items()):
print(f" {zone}: {count}")
# Validate
print("\n Validating...")
errors = validate(conn)
if errors:
print(f" ERRORS ({len(errors)}):")
for e in errors:
# 6. Gate energy connectivity (D-186) — must run after currency zones
print(" [6/8] Setting gate energy connectivity...")
energy = set_gate_energy(conn, args.dry_run)
for label, count in sorted(energy.items()):
print(f" {label}: {count}")
# 7. Sync corporations from wiki (D-182: hard error on name divergence)
print(" [7/8] Syncing corporations...")
corp_errors = sync_corporations(conn, wiki_corps, args.dry_run)
if corp_errors:
print(f" ERRORS ({len(corp_errors)}) — name divergence detected (D-182):")
for e in corp_errors:
print(f" - {e}")
print(" Fix: update wiki title or DB proper_name to match, then re-run.")
conn.close()
sys.exit(1)
n_db_corps = conn.execute("SELECT COUNT(*) FROM corporations").fetchone()[0]
print(f" {n_db_corps} corporations in DB ({len(wiki_corps)} from wiki)")
# 8. Corp presence from wiki headquarters data
print(" [8/8] Importing corp presence...")
commodity_ids = {
r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall()
}
n_presence = import_corp_presence(conn, wiki_corps, commodity_ids, args.dry_run)
print(f" {n_presence} corp_presence rows")
# Validate structural integrity (FK, chain refs, chain completeness).
# These errors indicate broken imported data — do NOT commit.
print("\n Validating structural integrity...")
struct_errors = validate(conn)
if struct_errors:
print(f" STRUCTURAL ERRORS ({len(struct_errors)}) — rolling back:")
for e in struct_errors:
print(f" - {e}")
conn.close()
sys.exit(1)
else:
print(" FK integrity OK")
print(" FK integrity and chain completeness OK")
# Commit all imported data (corps, presence, etc.) before coverage check.
# Coverage validation is a Phase 2 gate (D-175) — data should be persisted
# so tools can query it and report gaps clearly.
if not args.dry_run:
conn.commit()
print("\n Committed.")
print(" Data committed.")
else:
print("\n Dry run — no changes written.")
print(" Dry run — no changes written.")
# Validate coverage (hard errors per D-175, but after commit so data is usable).
print("\n Validating coverage (D-175 Phase 2 gate)...")
coverage_errors: list[str] = []
commodity_ids_for_coverage = {
r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall()
}
coverage_errors.extend(
_validate_commodity_coverage(conn, wiki_corps, commodity_ids_for_coverage)
)
coverage_errors.extend(_validate_system_coverage(conn, wiki_corps))
if coverage_errors:
print(f" COVERAGE ERRORS ({len(coverage_errors)}) — Phase 2 gate not met:")
for e in coverage_errors:
print(f" - {e}")
print("\n Data committed but Phase 2 gate is NOT met. "
"Add corporations to meet coverage thresholds and re-run.")
conn.close()
sys.exit(1)
else:
print(" All coverage thresholds met — Phase 2 gate PASSED.")
conn.close()
print(f"\n Done: {n_links} gate_links, {n_commodities} commodities, "
f"{n_chains} chains, {n_inputs} inputs\n")
f"{n_chains} chains, {n_inputs} inputs, {n_presence} corp_presence\n")
if __name__ == "__main__":