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

feat(economics): import_system_specialization() — D-237 authored layer (#1013)

Browse Source 
Wires the D-237 authored specialization layer into import_economics.py:

- import_system_specialization(): reads specialization_vocabulary.toml
  (FK-validated against commodities; repopulates specialization_vocabulary)
  and system_specialization.toml (UPSERTs economic_specialization +
  cultural_specialization onto system_economy, dominant_faction onto
  system_factions for authored systems only). Hard errors on bad commodity
  FK / projected enum / unknown system_id abort the transaction; prints a
  coverage report + missing-row warnings.
- Called as step 4b in main() (after commodities so the FK resolves, before
  currency zones). economic/cultural columns are importer-owned and cleared
  to NULL first for idempotency; dominant_faction only overwritten for
  authored systems (never globally cleared — shared with other derivation).
- specialization_vocabulary added to MIGRATION_SQL (after commodities for FK)
  so the migration path on existing DBs creates the table, not just fresh
  systems-schema.sql builds — this is what #1011 missed. Also cleared before
  commodities in the FK-safe clear block.
- Both source TOMLs added to IMPORT_ECONOMICS_SOURCES and mirrored in
  check-systems-db-stamp GENERATOR_SOURCES so editing them flips the stamp.

Validated: full non-dry-run import on a copy of the live systems.db exits 0;
27 vocab rows, 27/27/27 economic/cultural/faction set, no missing rows;
spot checks Ran=breadbasket/agrarian, Vuurkloof=independent,
financial_hub=NonPhysical/Specialist. Dry-run also exits 0.

V-SES CI guardrail suite remains #1015; code note flags that V-SES-03
(equal-or-higher) must not hard-fail HUB specializations.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-06-01 09:09:06 +02:00
co-authored by Claude Opus 4.8
parent 4f53793755
commit 883c1ab0b1
2 changed files with 221 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tooling/economy-db/import_economics.py
+217
View File
@@ -43,6 +43,8 @@ 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"
SPECIALIZATION_VOCAB_TOML = REPO_ROOT / "wiki" / "economics" / "specialization_vocabulary.toml"
SYSTEM_SPECIALIZATION_TOML = REPO_ROOT / "wiki" / "economics" / "system_specialization.toml"
SCHEMA_SQL = REPO_ROOT / "server" / "data" / "systems-schema.sql"
CORPORATIONS_DIR = REPO_ROOT / "wiki" / "corporations"
WIKI_STAR_SYSTEMS = REPO_ROOT / "wiki" / "star-systems"
@@ -83,6 +85,11 @@ IMPORT_ECONOMICS_SOURCES: tuple[Path, ...] = (
GENERATE_BRANDS_NAMES_RS,
GENERATE_BRANDS_WRAPPER,
REPO_ROOT / "tooling" / "schema_version.py",
# D-237 authored specialization layer: these data TOMLs feed the DB, so a
# change to either must flip the stamp and force a regen (#1013). Mirror in
# GENERATOR_SOURCES["import_economics"] in tooling/check-systems-db-stamp.
SPECIALIZATION_VOCAB_TOML,
SYSTEM_SPECIALIZATION_TOML,
)
@@ -240,6 +247,19 @@ CREATE TABLE IF NOT EXISTS commodities (
updated_at TEXT DEFAULT (datetime('now'))
);
-- D-237 authored specialization layer vocabulary (must follow commodities for FK).
-- Mirrors the canonical DDL in systems-schema.sql; here so the migration path
-- (existing DBs) gets the table, not just fresh systems-schema.sql builds.
CREATE TABLE IF NOT EXISTS specialization_vocabulary (
specialization_id TEXT PRIMARY KEY,
commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
production_ubiquity_override TEXT,
bulk_class_projected TEXT NOT NULL,
production_ubiquity_projected TEXT NOT NULL,
description TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_spec_vocab_commodity ON specialization_vocabulary(commodity_id);
CREATE TABLE IF NOT EXISTS production_chains (
chain_id TEXT PRIMARY KEY,
output_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
@@ -1571,6 +1591,178 @@ def validate_brands(conn: sqlite3.Connection) -> list[str]:
return errors
# ---------------------------------------------------------------------------
# System specialization — D-237 authored layer
# ---------------------------------------------------------------------------
# Authoritative D-233 projected enums. The full CI guardrail suite (V-SES-*)
# lands in #1015; this function performs the FK + enum sanity the import itself
# needs to stay sound. NOTE for #1015: the D-237 "equal-or-higher" override rule
# must NOT hard-fail HUB specializations (shipbuilding, transit_hub) whose local
# production_ubiquity_projected is intentionally below their commodity's global
# default — see specialization_vocabulary.toml header.
_BULK_CLASSES = {"BulkSolid", "BulkLiquid", "PrecisionDense", "Perishable", "NonPhysical"}
_PRODUCTION_UBIQUITY = {"Ubiquitous", "Common", "Specialist", "MonopolySource"}
_FACTION_VOCAB = {
"concord_assembly", "compact", "compact_sympathetic", "syndic_dominant",
"veil_institute", "independent", "disputed", "mixed",
}
def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dict:
"""Import the D-237 authored specialization layer.
Reads two TOMLs:
- specialization_vocabulary.toml -> specialization_vocabulary table
(FK-validated against commodities; MUST run after import_commodities).
- system_specialization.toml -> UPSERTs economic_specialization +
cultural_specialization onto system_economy, and dominant_faction onto
system_factions, for authored (hero) systems only.
Authored lore wins; unauthored systems are left NULL for the generator's
heuristic fallback. economic_specialization / cultural_specialization are
owned exclusively by this importer, so they are cleared to NULL first for
idempotency (a removed stanza must not leave a stale value). dominant_faction
is shared with other derivation paths, so it is ONLY overwritten for systems
present in the TOML (per #1013) — never globally cleared.
Returns a coverage dict for the caller's report. Raises _ImportAborted on a
hard validation failure (FK, enum, or unknown system_id).
"""
with open(SPECIALIZATION_VOCAB_TOML, "rb") as f:
vocab = tomllib.load(f)
with open(SYSTEM_SPECIALIZATION_TOML, "rb") as f:
systems = tomllib.load(f)
commodity_ids = {
r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall()
}
system_ids = {
r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
}
economy_system_ids = {
r[0] for r in conn.execute("SELECT system_id FROM system_economy").fetchall()
}
faction_system_ids = {
r[0] for r in conn.execute("SELECT system_id FROM system_factions").fetchall()
}
errors: list[str] = []
# --- Vocabulary: FK + enum validation -------------------------------
vocab_rows = []
for spec_id, v in vocab.items():
cid = v.get("commodity_id")
if cid not in commodity_ids:
errors.append(
f"specialization_vocabulary '{spec_id}': commodity_id "
f"'{cid}' not in commodities catalog"
)
bc = v.get("bulk_class_projected")
if bc not in _BULK_CLASSES:
errors.append(
f"specialization_vocabulary '{spec_id}': bulk_class_projected "
f"'{bc}' invalid (expected one of {sorted(_BULK_CLASSES)})"
)
pu = v.get("production_ubiquity_projected")
if pu not in _PRODUCTION_UBIQUITY:
errors.append(
f"specialization_vocabulary '{spec_id}': "
f"production_ubiquity_projected '{pu}' invalid "
f"(expected one of {sorted(_PRODUCTION_UBIQUITY)})"
)
override = v.get("production_ubiquity_override") or None # "" -> NULL
vocab_rows.append((spec_id, cid, override, bc, pu, v.get("description", "")))
valid_spec_ids = set(vocab.keys())
# --- System stanzas: id + value validation --------------------------
for sid, s in systems.items():
if sid not in system_ids:
errors.append(
f"system_specialization '{sid}': not a known star_systems.system_id"
)
es = s.get("economic_specialization")
if es is not None and es not in valid_spec_ids:
errors.append(
f"system_specialization '{sid}': economic_specialization "
f"'{es}' not in specialization_vocabulary"
)
df = s.get("dominant_faction")
if df is not None and df not in _FACTION_VOCAB:
errors.append(
f"system_specialization '{sid}': dominant_faction "
f"'{df}' invalid (expected one of {sorted(_FACTION_VOCAB)})"
)
if errors:
print(f" SPECIALIZATION ERRORS ({len(errors)}):")
for e in errors:
print(f" - {e}")
raise _ImportAborted()
coverage = {
"vocab": len(vocab_rows),
"economic": 0,
"cultural": 0,
"faction": 0,
"missing_economy_row": [],
"missing_faction_row": [],
}
if dry_run:
for sid, s in systems.items():
coverage["economic"] += 1 if s.get("economic_specialization") else 0
coverage["cultural"] += 1 if s.get("cultural_specialization") else 0
coverage["faction"] += 1 if s.get("dominant_faction") else 0
return coverage
# --- Repopulate vocabulary table ------------------------------------
conn.execute("DELETE FROM specialization_vocabulary")
conn.executemany(
"""INSERT INTO specialization_vocabulary (
specialization_id, commodity_id, production_ubiquity_override,
bulk_class_projected, production_ubiquity_projected, description
) VALUES (?, ?, ?, ?, ?, ?)""",
vocab_rows,
)
# --- Clear importer-owned columns (idempotency) ---------------------
conn.execute(
"UPDATE system_economy SET economic_specialization = NULL, "
"cultural_specialization = NULL"
)
# --- UPSERT per-system authored fields ------------------------------
for sid, s in systems.items():
es = s.get("economic_specialization")
cs = s.get("cultural_specialization")
if sid in economy_system_ids:
conn.execute(
"UPDATE system_economy SET economic_specialization = ?, "
"cultural_specialization = ? WHERE system_id = ?",
(es, cs, sid),
)
coverage["economic"] += 1 if es else 0
coverage["cultural"] += 1 if cs else 0
else:
coverage["missing_economy_row"].append(sid)
df = s.get("dominant_faction")
if df is not None:
if sid in faction_system_ids:
conn.execute(
"UPDATE system_factions SET dominant_faction = ? "
"WHERE system_id = ?",
(df, sid),
)
coverage["faction"] += 1
else:
coverage["missing_faction_row"].append(sid)
return coverage
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
@@ -1642,6 +1834,9 @@ def main():
conn.execute("DELETE FROM corp_lifecycle_events")
conn.execute("DELETE FROM chain_inputs")
conn.execute("DELETE FROM production_chains")
# specialization_vocabulary FK-references commodities — clear it
# before commodities so the FK-on delete does not fail (D-237).
conn.execute("DELETE FROM specialization_vocabulary")
conn.execute("DELETE FROM commodities")
conn.execute("DELETE FROM gate_links")
@@ -1660,6 +1855,28 @@ def main():
n_chains, n_inputs = import_chains(conn, args.dry_run)
print(f" {n_chains} chains, {n_inputs} inputs")
# 4b. System specialization (D-237 authored layer) — after commodities
# (FK) and before currency zones. UPSERTs onto pre-existing
# system_economy / system_factions rows; unauthored systems stay NULL.
print(" [4b/10] Importing system specialization (D-237)...")
spec = import_system_specialization(conn, args.dry_run)
print(
f" vocab {spec['vocab']} | economic {spec['economic']} | "
f"cultural {spec['cultural']} | faction {spec['faction']}"
)
if spec["missing_economy_row"]:
print(
f" WARNING: {len(spec['missing_economy_row'])} authored "
f"system(s) lack a system_economy row (values dropped): "
f"{spec['missing_economy_row']}"
)
if spec["missing_faction_row"]:
print(
f" WARNING: {len(spec['missing_faction_row'])} authored "
f"system(s) lack a system_factions row (faction dropped): "
f"{spec['missing_faction_row']}"
)
# 5. Currency zones
print(" [5/10] Setting currency zones...")
zones = set_currency_zones(conn, args.dry_run)