388 lines
13 KiB
Python
Executable File
388 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
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)
|
|
- production_chains + chain_inputs from wiki/economics/production_chains.toml
|
|
- currency_zone on star_systems (default TRACTUS_PRIMARY)
|
|
|
|
Does NOT populate corp_presence — that's a future pipeline step.
|
|
|
|
Usage:
|
|
python3 tooling/economy-db/import_economics.py
|
|
python3 tooling/economy-db/import_economics.py --dry-run
|
|
python3 tooling/economy-db/import_economics.py --db path/to/systems.db
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
|
|
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
|
|
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"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schema migration — add new tables and columns to existing DB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MIGRATION_SQL = """
|
|
-- Economics tables (idempotent — safe to re-run)
|
|
|
|
CREATE TABLE IF NOT EXISTS gate_links (
|
|
from_system_id TEXT NOT NULL REFERENCES star_systems(system_id),
|
|
to_system_id TEXT NOT NULL REFERENCES star_systems(system_id),
|
|
PRIMARY KEY (from_system_id, to_system_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS commodities (
|
|
commodity_id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
tier TEXT NOT NULL,
|
|
elasticity TEXT NOT NULL,
|
|
base_price REAL NOT NULL,
|
|
bulk_class TEXT,
|
|
unit TEXT,
|
|
production_ubiquity TEXT,
|
|
demand_model TEXT,
|
|
commission_certifiable INTEGER DEFAULT 0,
|
|
compact_contested INTEGER DEFAULT 0,
|
|
shadow_viable INTEGER DEFAULT 0,
|
|
panic_threshold_weeks INTEGER DEFAULT 0,
|
|
description TEXT,
|
|
updated_at TEXT DEFAULT (datetime('now'))
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS production_chains (
|
|
chain_id TEXT PRIMARY KEY,
|
|
output_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
|
|
output_quantity REAL NOT NULL DEFAULT 1.0,
|
|
location_bound INTEGER DEFAULT 0,
|
|
description TEXT,
|
|
updated_at TEXT DEFAULT (datetime('now'))
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS chain_inputs (
|
|
chain_id TEXT NOT NULL REFERENCES production_chains(chain_id),
|
|
input_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
|
|
quantity REAL NOT NULL,
|
|
PRIMARY KEY (chain_id, input_commodity_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS corp_presence (
|
|
corp_id TEXT NOT NULL REFERENCES corporations(corp_id),
|
|
location_id TEXT NOT NULL,
|
|
location_type TEXT NOT NULL,
|
|
primary_operation TEXT,
|
|
updated_at TEXT DEFAULT (datetime('now')),
|
|
PRIMARY KEY (corp_id, location_id)
|
|
);
|
|
|
|
-- New indexes
|
|
CREATE INDEX IF NOT EXISTS idx_star_systems_currency ON star_systems(currency_zone);
|
|
CREATE INDEX IF NOT EXISTS idx_gate_links_from ON gate_links(from_system_id);
|
|
CREATE INDEX IF NOT EXISTS idx_gate_links_to ON gate_links(to_system_id);
|
|
CREATE INDEX IF NOT EXISTS idx_commodities_tier ON commodities(tier);
|
|
CREATE INDEX IF NOT EXISTS idx_production_chains_output ON production_chains(output_commodity_id);
|
|
CREATE INDEX IF NOT EXISTS idx_chain_inputs_commodity ON chain_inputs(input_commodity_id);
|
|
CREATE INDEX IF NOT EXISTS idx_corp_presence_corp ON corp_presence(corp_id);
|
|
CREATE INDEX IF NOT EXISTS idx_corp_presence_location ON corp_presence(location_id);
|
|
"""
|
|
|
|
# Columns to add to existing tables (ALTER TABLE is idempotent via try/except)
|
|
COLUMN_MIGRATIONS = [
|
|
("star_systems", "currency_zone", "TEXT DEFAULT 'TRACTUS_PRIMARY'"),
|
|
("corporations", "behavioral_archetype", "TEXT"),
|
|
("corporations", "supply_chain_role", "TEXT"),
|
|
("corporations", "shadow_economy_access", "INTEGER DEFAULT 0"),
|
|
]
|
|
|
|
|
|
def _add_column(conn: sqlite3.Connection, table: str, col: str, col_type: str):
|
|
"""Add a column if it doesn't exist. SQLite has no IF NOT EXISTS for ALTER."""
|
|
try:
|
|
conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {col_type}")
|
|
except sqlite3.OperationalError as e:
|
|
if "duplicate column" in str(e).lower():
|
|
pass # already exists
|
|
else:
|
|
raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gate links
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def import_gate_links(conn: sqlite3.Connection, dry_run: bool) -> int:
|
|
with open(STAR_MAP) as f:
|
|
data = json.load(f)
|
|
|
|
edges = data["edges"]
|
|
system_ids = {r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()}
|
|
|
|
rows = []
|
|
skipped = []
|
|
for a, b in edges:
|
|
if a not in system_ids:
|
|
skipped.append(a)
|
|
continue
|
|
if b not in system_ids:
|
|
skipped.append(b)
|
|
continue
|
|
rows.append((a, b))
|
|
rows.append((b, a))
|
|
|
|
if skipped:
|
|
unique_skipped = sorted(set(skipped))
|
|
print(f" warning: {len(unique_skipped)} system(s) in star-map.json not in DB: {unique_skipped[:5]}...")
|
|
|
|
if not dry_run:
|
|
conn.executemany(
|
|
"INSERT OR IGNORE INTO gate_links (from_system_id, to_system_id) VALUES (?, ?)",
|
|
rows,
|
|
)
|
|
|
|
return len(rows)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commodities
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def import_commodities(conn: sqlite3.Connection, dry_run: bool) -> int:
|
|
with open(COMMODITIES_TOML, "rb") as f:
|
|
data = tomllib.load(f)
|
|
|
|
rows = []
|
|
for cid, c in data.items():
|
|
rows.append((
|
|
cid,
|
|
c["name"],
|
|
c["tier"],
|
|
c["elasticity"],
|
|
c["base_price"],
|
|
c.get("bulk_class"),
|
|
c.get("unit"),
|
|
c.get("production_ubiquity"),
|
|
c.get("demand_model"),
|
|
int(c.get("commission_certifiable", False)),
|
|
int(c.get("compact_contested", False)),
|
|
int(c.get("shadow_viable", False)),
|
|
c.get("panic_threshold_weeks", 0),
|
|
c.get("description"),
|
|
))
|
|
|
|
if not dry_run:
|
|
conn.executemany(
|
|
"""INSERT INTO commodities (
|
|
commodity_id, name, tier, elasticity, base_price,
|
|
bulk_class, unit, production_ubiquity, demand_model,
|
|
commission_certifiable, compact_contested, shadow_viable,
|
|
panic_threshold_weeks, description
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
rows,
|
|
)
|
|
|
|
return len(rows)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Production chains + inputs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def import_chains(conn: sqlite3.Connection, dry_run: bool) -> tuple[int, int]:
|
|
with open(CHAINS_TOML, "rb") as f:
|
|
data = tomllib.load(f)
|
|
|
|
chain_rows = []
|
|
input_rows = []
|
|
for chain_id, c in data.items():
|
|
chain_rows.append((
|
|
chain_id,
|
|
c["output"],
|
|
c.get("output_quantity", 1.0),
|
|
int(c.get("location_bound", False)),
|
|
c.get("description"),
|
|
))
|
|
for inp in c.get("inputs", []):
|
|
input_rows.append((
|
|
chain_id,
|
|
inp["commodity"],
|
|
inp["quantity"],
|
|
))
|
|
|
|
if not dry_run:
|
|
conn.executemany(
|
|
"""INSERT INTO production_chains (
|
|
chain_id, output_commodity_id, output_quantity,
|
|
location_bound, description
|
|
) VALUES (?, ?, ?, ?, ?)""",
|
|
chain_rows,
|
|
)
|
|
conn.executemany(
|
|
"""INSERT INTO chain_inputs (
|
|
chain_id, input_commodity_id, quantity
|
|
) VALUES (?, ?, ?)""",
|
|
input_rows,
|
|
)
|
|
|
|
return len(chain_rows), len(input_rows)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Currency zones
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def set_currency_zones(conn: sqlite3.Connection, dry_run: bool) -> dict:
|
|
"""Set currency_zone on star_systems. Default TRACTUS_PRIMARY, Sol = MIXED."""
|
|
if dry_run:
|
|
return {"TRACTUS_PRIMARY": "all", "MIXED": "GJ 0"}
|
|
|
|
# Default everything to TRACTUS_PRIMARY
|
|
conn.execute("UPDATE star_systems SET currency_zone = 'TRACTUS_PRIMARY' WHERE currency_zone IS NULL")
|
|
|
|
# Sol system is MIXED (Earth legacy currency presence)
|
|
conn.execute("UPDATE star_systems SET currency_zone = 'MIXED' WHERE system_id = 'GJ 0'")
|
|
|
|
# Future: Compact systems → MARK_PRIMARY (requires authored Compact membership data)
|
|
|
|
counts = {}
|
|
for row in conn.execute("SELECT currency_zone, COUNT(*) FROM star_systems GROUP BY currency_zone"):
|
|
counts[row[0]] = row[1]
|
|
|
|
return counts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def validate(conn: sqlite3.Connection) -> list[str]:
|
|
errors = []
|
|
|
|
# FK integrity
|
|
fk_issues = conn.execute("PRAGMA foreign_key_check").fetchall()
|
|
if fk_issues:
|
|
for issue in fk_issues[:10]:
|
|
errors.append(f"FK violation: table={issue[0]} rowid={issue[1]} "
|
|
f"parent={issue[2]} fkid={issue[3]}")
|
|
|
|
# Chain inputs reference valid commodities
|
|
orphan_inputs = conn.execute("""
|
|
SELECT ci.chain_id, ci.input_commodity_id
|
|
FROM chain_inputs ci
|
|
LEFT JOIN commodities c ON ci.input_commodity_id = c.commodity_id
|
|
WHERE c.commodity_id IS NULL
|
|
""").fetchall()
|
|
for chain_id, cid in orphan_inputs:
|
|
errors.append(f"chain_inputs: chain '{chain_id}' references unknown commodity '{cid}'")
|
|
|
|
# Chain outputs reference valid commodities
|
|
orphan_outputs = conn.execute("""
|
|
SELECT pc.chain_id, pc.output_commodity_id
|
|
FROM production_chains pc
|
|
LEFT JOIN commodities c ON pc.output_commodity_id = c.commodity_id
|
|
WHERE c.commodity_id IS NULL
|
|
""").fetchall()
|
|
for chain_id, cid in orphan_outputs:
|
|
errors.append(f"production_chains: chain '{chain_id}' outputs unknown commodity '{cid}'")
|
|
|
|
return errors
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Import economics data into systems.db")
|
|
parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db")
|
|
parser.add_argument("--dry-run", action="store_true", help="Validate without writing")
|
|
args = parser.parse_args()
|
|
|
|
db_path = Path(args.db)
|
|
if not db_path.exists():
|
|
print(f"error: {db_path} not found", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"\n Economics Import Pipeline")
|
|
print(f" DB: {db_path}")
|
|
if args.dry_run:
|
|
print(f" Mode: DRY RUN")
|
|
print()
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
|
|
# 1. Migrate schema
|
|
print(" [1/5] 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)
|
|
if not args.dry_run:
|
|
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...")
|
|
n_links = import_gate_links(conn, args.dry_run)
|
|
print(f" {n_links} rows (bidirectional)")
|
|
|
|
# 3. Commodities
|
|
print(" [3/5] Importing commodities...")
|
|
n_commodities = import_commodities(conn, args.dry_run)
|
|
print(f" {n_commodities} commodities")
|
|
|
|
# 4. Production chains
|
|
print(" [4/5] 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...")
|
|
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:
|
|
print(f" - {e}")
|
|
conn.close()
|
|
sys.exit(1)
|
|
else:
|
|
print(" FK integrity OK")
|
|
|
|
if not args.dry_run:
|
|
conn.commit()
|
|
print("\n Committed.")
|
|
else:
|
|
print("\n Dry run — no changes written.")
|
|
|
|
conn.close()
|
|
|
|
print(f"\n Done: {n_links} gate_links, {n_commodities} commodities, "
|
|
f"{n_chains} chains, {n_inputs} inputs\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|