refactor(tooling): extract atlas_common from generate_atlas (D-223 #951)
Move the shared atlas-DB utilities (schema application, inhabited-body query, body-def loader, grid constants) out of the soon-to-be-retired generate_atlas.py into a dedicated atlas_common.py with no dependency on geometry-production code. Repoint the surviving build-time importers (import_heightmaps, import_province_boundaries) at the new module. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.7
@@ -0,0 +1,164 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
atlas_common.py — shared atlas-DB utilities for the planet-gen importers.
|
||||
|
||||
Extracted from the retired generate_atlas.py (D-223, #951). The atlas city/road/
|
||||
river geometry *generator* was retired when authored geometry was dropped in
|
||||
favour of the deterministic server-side cascade (Phase 4); markers.json is now a
|
||||
flavoured name pool only. What remained were the schema/DB helpers shared by the
|
||||
surviving build-time importers (heightmaps, province boundaries, city names) and
|
||||
the naming pipeline. Those live here so there is a single home for them with no
|
||||
dependency on the deleted geometry-production code.
|
||||
|
||||
This is a library module — it does NOT re-exec the venv. The importing scripts
|
||||
(import_heightmaps.py, import_province_boundaries.py, import_city_names.py,
|
||||
gemma_naming.py) own the venv bootstrap before importing this module.
|
||||
|
||||
Decisions: D-223 (authored content as flavoured name pool), D-191 (atlas index).
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paths and grid constants
|
||||
# ---------------------------------------------------------------------------
|
||||
TOOLING_DIR = Path(__file__).resolve().parent
|
||||
REPO_ROOT = (TOOLING_DIR / ".." / "..").resolve()
|
||||
|
||||
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
|
||||
WIKI_SYSTEMS = REPO_ROOT / "wiki" / "star-systems"
|
||||
SYSTEMS_SCHEMA_PATH = REPO_ROOT / "server" / "data" / "systems-schema.sql"
|
||||
|
||||
# Heightmap pixel-space grid (D-202). Geometry derives from this resolution.
|
||||
GRID_W = 512
|
||||
GRID_H = 256
|
||||
|
||||
# Delimiters for the single canonical atlas_* DDL block in
|
||||
# server/data/systems-schema.sql. `_load_atlas_schema()` extracts everything
|
||||
# between these markers at runtime so this file does not duplicate the schema
|
||||
# (and drift from it).
|
||||
_ATLAS_SCHEMA_BEGIN_MARKER = "-- BEGIN ATLAS INDEX"
|
||||
_ATLAS_SCHEMA_END_MARKER = "-- END ATLAS INDEX"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Schema
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _load_atlas_schema() -> str:
|
||||
"""Return the atlas_* DDL block from systems-schema.sql.
|
||||
|
||||
systems-schema.sql is the single source of truth for the atlas index
|
||||
tables (see the BEGIN ATLAS INDEX / END ATLAS INDEX markers). We extract
|
||||
just that block and run it through `executescript` so callers work against
|
||||
any DB state (fresh or partially-migrated) without maintaining a second
|
||||
copy of the DDL here.
|
||||
"""
|
||||
if not SYSTEMS_SCHEMA_PATH.exists():
|
||||
raise RuntimeError(
|
||||
f"systems-schema.sql not found at {SYSTEMS_SCHEMA_PATH} — "
|
||||
"atlas importers cannot proceed without the canonical schema."
|
||||
)
|
||||
text = SYSTEMS_SCHEMA_PATH.read_text()
|
||||
try:
|
||||
start = text.index(_ATLAS_SCHEMA_BEGIN_MARKER)
|
||||
end = text.index(_ATLAS_SCHEMA_END_MARKER, start)
|
||||
except ValueError as e:
|
||||
raise RuntimeError(
|
||||
f"systems-schema.sql is missing the {_ATLAS_SCHEMA_BEGIN_MARKER}/"
|
||||
f"{_ATLAS_SCHEMA_END_MARKER} block — has the schema been "
|
||||
"restructured?"
|
||||
) from e
|
||||
return text[start:end]
|
||||
|
||||
|
||||
def ensure_atlas_schema(conn: sqlite3.Connection) -> None:
|
||||
"""Apply the canonical atlas_* DDL from systems-schema.sql.
|
||||
|
||||
Idempotent: all statements inside the block use CREATE TABLE / INDEX
|
||||
IF NOT EXISTS, so running this on an already-migrated DB is a no-op.
|
||||
|
||||
Also ensures the meta stamp table exists (#855, #856).
|
||||
"""
|
||||
conn.executescript(_load_atlas_schema())
|
||||
conn.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS meta (
|
||||
generator_name TEXT PRIMARY KEY,
|
||||
schema_version TEXT NOT NULL,
|
||||
schema_sha TEXT,
|
||||
generator_sha TEXT NOT NULL,
|
||||
generated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
"""
|
||||
)
|
||||
# Add schema_sha column to existing DBs that pre-date #888 (#888 migration).
|
||||
try:
|
||||
conn.execute("ALTER TABLE meta ADD COLUMN schema_sha TEXT")
|
||||
except sqlite3.OperationalError as e:
|
||||
if "duplicate column" not in str(e).lower():
|
||||
raise
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Body definition loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def load_body_def(body_dir: Path) -> dict | None:
|
||||
"""Load body definition from body index.md frontmatter."""
|
||||
index_md = body_dir / "index.md"
|
||||
if not index_md.exists():
|
||||
return None
|
||||
content = index_md.read_text()
|
||||
if not content.startswith("---"):
|
||||
return None
|
||||
try:
|
||||
end = content.index("---", 3)
|
||||
bd = yaml.safe_load(content[3:end])
|
||||
except (ValueError, yaml.YAMLError):
|
||||
return None
|
||||
if not bd or "id" not in bd or "planet_class" not in bd:
|
||||
return None
|
||||
return bd
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DB queries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def query_inhabited_bodies(conn: sqlite3.Connection) -> list[dict]:
|
||||
"""Query systems.db for all inhabited bodies with terrain_reference set."""
|
||||
rows = conn.execute("""
|
||||
SELECT
|
||||
b.body_id,
|
||||
b.system_id,
|
||||
b.terrain_reference,
|
||||
b.population,
|
||||
b.settlement_pattern,
|
||||
b.planet_class,
|
||||
b.economic_role,
|
||||
COALESCE(b.cultural_corridor, s.cultural_corridor) AS cultural_corridor
|
||||
FROM bodies b
|
||||
JOIN star_systems s ON b.system_id = s.system_id
|
||||
WHERE b.inhabited = 1
|
||||
AND b.terrain_reference IS NOT NULL
|
||||
AND b.population > 0
|
||||
ORDER BY b.system_id, b.body_id
|
||||
""").fetchall()
|
||||
|
||||
return [
|
||||
{
|
||||
"body_id": row[0],
|
||||
"system_id": row[1],
|
||||
"terrain_reference": row[2],
|
||||
"population": row[3] or 0,
|
||||
"settlement_pattern": row[4],
|
||||
"planet_class": row[5],
|
||||
"economic_role": row[6],
|
||||
"cultural_corridor": row[7],
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
@@ -41,7 +41,7 @@ if _venv_python.exists() and Path(sys.executable).resolve() != _venv_python.reso
|
||||
import numpy as np
|
||||
import sqlite3
|
||||
|
||||
from generate_atlas import (
|
||||
from atlas_common import (
|
||||
GRID_W,
|
||||
GRID_H,
|
||||
DB_PATH,
|
||||
|
||||
@@ -59,7 +59,7 @@ if _venv_python.exists() and Path(sys.executable).resolve() != _venv_python.reso
|
||||
import numpy as np
|
||||
import sqlite3
|
||||
|
||||
from generate_atlas import (
|
||||
from atlas_common import (
|
||||
DB_PATH,
|
||||
ensure_atlas_schema,
|
||||
query_inhabited_bodies,
|
||||
|
||||
Reference in New Issue
Block a user