feat(tooling): Gemma 4 batch naming pipeline with wiki-grounded register selection (#833)
Replace the one-at-a-time Gemma 2 naming pipeline with a batch-oriented Gemma 4 E2B pipeline. Key changes: - naming_core.py: shared library with Levenshtein distinctiveness ranking, batch prompt building, mood injection pool, name validation, and adjacent-register refill logic - Wiki-grounded register selection: per-system LLM call picks the cultural register based on wiki/GTTR content instead of hash randomizer - Batch naming: requests N*2 names per call, ranks by word-average Levenshtein distance, fills quota from most-distinct candidates - Mood pool: 13 emotional seeds randomized per-body for vocabulary divergence (ambition, fear, isolation, defiance, etc.) - Adjacent-register refill: when primary register exhausts, automatically switches to next corridor substyle - Inhabited-first body ordering: habitable worlds get first pick of register vocabulary, barren moons get leftovers - Process group cleanup: SIGTERM/SIGKILL the full distrobox chain on subprocess refresh to prevent GPU zombie processes - qa_naming.py: QA report, fix_fewshot_bleed.py: post-hoc fix script - test_batch_naming.py, test_register_selection.py: test harnesses Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,246 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replace few-shot example names that bled into the output.
|
||||
|
||||
The batch naming prompt uses Scottish Highland and Dutch colonial
|
||||
examples. The Scottish ones (Glen Moray, Dunvegan Ridge, Torridon,
|
||||
Cairn Brae, The Kelpie's Spine) leaked into 270 features. This script
|
||||
replaces them with unique names from a combined Scottish/Welsh/Irish
|
||||
pool, ensuring no collisions with the existing corpus.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
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"
|
||||
|
||||
sys.path.insert(0, str(TOOLING_DIR))
|
||||
from generate_atlas import sync_markers_to_db
|
||||
|
||||
# The few-shot names to replace
|
||||
FEWSHOT_NAMES = {
|
||||
"glen moray", "dunvegan ridge", "torridon", "cairn brae", "the kelpie's spine",
|
||||
"kloosterbeek", "nieuw rijn", "hoogland run", "van diemen's creek",
|
||||
}
|
||||
|
||||
# Scottish / Welsh / Irish replacement pool — 300+ names to cover 270 replacements
|
||||
# with room for Levenshtein filtering. Mix of geographic feature styles.
|
||||
REPLACEMENT_POOL = [
|
||||
# Scottish
|
||||
"Glenfinnan", "Dalwhinnie Pass", "Cairngorm", "Loch Maree",
|
||||
"Kinlochleven", "Strathspey", "Brae Morar", "Skye Reach",
|
||||
"Ardnamurchan", "Kintail", "Glen Affric", "Lochaber",
|
||||
"Killiecrankie", "Rannoch Moor", "Glen Coe", "Strathnaver",
|
||||
"Applecross", "Torrisdale", "Durness", "Assynt",
|
||||
"Coigach", "Inverpolly", "Sandwood", "Cape Wrath",
|
||||
"Sutherland", "Helmsdale", "Brora", "Golspie",
|
||||
"Cromarty", "Dornoch", "Nairn", "Forres",
|
||||
"Culbin", "Findhorn", "Spey Bay", "Buckie",
|
||||
"Banff", "Fraserburgh", "Peterhead", "Cruden Bay",
|
||||
"Slains", "Ythan", "Bennachie", "Morven",
|
||||
"Lochnagar", "Braemar", "Balmoral", "Crathie",
|
||||
"Ballater", "Dinnet", "Tarland", "Lumphanan",
|
||||
"Corgarff", "Tomintoul", "Glenlivet", "Dufftown",
|
||||
"Craigellachie", "Aberlour", "Knockando", "Archiestown",
|
||||
"Rothes", "Elgin", "Lossiemouth", "Burghead",
|
||||
"Kinloss", "Alves", "Pluscarden", "Dallas",
|
||||
# Welsh
|
||||
"Cwm Idwal", "Beddgelert", "Crib Goch", "Tryfan",
|
||||
"Ogwen", "Llyn Padarn", "Dolgellau", "Harlech",
|
||||
"Rhinog", "Cader Idris", "Barmouth", "Aberdovey",
|
||||
"Tywyn", "Machynlleth", "Pumlumon", "Hafren",
|
||||
"Elan Valley", "Claerwen", "Llandrindod", "Brecon",
|
||||
"Pen y Fan", "Corn Du", "Crickhowell", "Llangorse",
|
||||
"Talgarth", "Hay Bluff", "Mynydd Troed", "Mynydd Llangorse",
|
||||
"Skirrid", "Blorenge", "Llanfoist", "Govilon",
|
||||
"Gilwern", "Llangattock", "Crug Hywel", "Cwm Clydach",
|
||||
"Pontneddfechan", "Ystradfellte", "Sgwd yr Eira", "Henrhyd",
|
||||
"Carreg Cennen", "Dinefwr", "Llandeilo", "Dryslwyn",
|
||||
"Tywi Valley", "Carmarthen", "Kidwelly", "Pembrey",
|
||||
"Gower", "Rhossili", "Oxwich", "Port Eynon",
|
||||
"Pennard", "Langland", "Caswell", "Mumbles",
|
||||
"Merthyr Mawr", "Ogmore", "Dunraven", "Llantwit",
|
||||
"Monknash", "Nash Point", "Aberthaw", "Fonmon",
|
||||
# Irish
|
||||
"Glendalough", "Lugnaquilla", "Glen Imaal", "Wicklow Gap",
|
||||
"Sally Gap", "Kippure", "Djuce", "Maulin",
|
||||
"Djouce", "Great Sugar Loaf", "Bray Head", "Killiney",
|
||||
"Dalkey", "Howth", "Lambay", "Ireland's Eye",
|
||||
"Malahide", "Portmarnock", "Donabate", "Skerries",
|
||||
"Balbriggan", "Gormanston", "Bettystown", "Laytown",
|
||||
"Slane", "Newgrange", "Dowth", "Knowth",
|
||||
"Tara", "Trim", "Navan", "Kells",
|
||||
"Loughcrew", "Oldcastle", "Castlepollard", "Fore",
|
||||
"Delvin", "Mullingar", "Kilbeggan", "Tullamore",
|
||||
"Clara", "Ferbane", "Banagher", "Shannonbridge",
|
||||
"Clonmacnoise", "Ballinasloe", "Aughrim", "Loughrea",
|
||||
"Portumna", "Mountshannon", "Killaloe", "Ballina",
|
||||
"Nenagh", "Roscrea", "Templemore", "Thurles",
|
||||
"Cashel", "Cahir", "Clonmel", "Carrick-on-Suir",
|
||||
"Piltown", "Mooncoin", "Waterford", "Tramore",
|
||||
"Bunmahon", "Ardmore", "Youghal", "Midleton",
|
||||
"Cobh", "Crosshaven", "Kinsale", "Clonakilty",
|
||||
"Skibbereen", "Bantry", "Glengarriff", "Kenmare",
|
||||
"Sneem", "Caherdaniel", "Waterville", "Cahersiveen",
|
||||
"Valentia", "Portmagee", "Skellig", "Dingle",
|
||||
"Brandon", "Castlegregory", "Fenit", "Tralee",
|
||||
"Listowel", "Ballybunion", "Tarbert", "Glin",
|
||||
"Foynes", "Askeaton", "Adare", "Patrickswell",
|
||||
# More Scottish/Gaelic to fill
|
||||
"Stornoway", "Tarbert", "Scalpay", "Eriskay",
|
||||
"Barra", "Vatersay", "Minguilay", "Pabbay",
|
||||
"Berneray", "Monach Isles", "Balranald", "Lochmaddy",
|
||||
"Benbecula", "Grimsay", "Ronay", "Wiay",
|
||||
"Canna", "Rum", "Eigg", "Muck",
|
||||
"Ardnish", "Arisaig", "Morar", "Mallaig",
|
||||
"Knoydart", "Barrisdale", "Arnisdale", "Glenelg",
|
||||
"Sandaig", "Brochs of Borve", "Callanish", "Garenin",
|
||||
"Carloway", "Arnol", "Barvas", "Tolsta",
|
||||
"Ness", "Europie", "Swainbost", "Skigersta",
|
||||
# Additional Welsh/Irish
|
||||
"Aberystwyth", "Llanberis", "Betws-y-Coed", "Conwy",
|
||||
"Caernarfon", "Pwllheli", "Abersoch", "Nefyn",
|
||||
"Llanbedrog", "Criccieth", "Porthmadog", "Portmeirion",
|
||||
"Trawsfynydd", "Ffestiniog", "Blaenau", "Llyn Tegid",
|
||||
"Corwen", "Llangollen", "Chirk", "Oswestry",
|
||||
]
|
||||
|
||||
|
||||
def load_global_names(conn):
|
||||
"""Load all existing names globally for uniqueness checking."""
|
||||
names = set()
|
||||
for table in ['atlas_cities', 'atlas_rivers', 'atlas_mountain_ranges',
|
||||
'atlas_oceans', 'atlas_pois']:
|
||||
rows = conn.execute(
|
||||
f"SELECT lower(name) FROM {table} WHERE name IS NOT NULL AND name != ''"
|
||||
).fetchall()
|
||||
names.update(r[0] for r in rows)
|
||||
return names
|
||||
|
||||
|
||||
def main():
|
||||
conn = sqlite3.connect(str(DB_PATH), timeout=30.0)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA busy_timeout=15000")
|
||||
|
||||
global_names = load_global_names(conn)
|
||||
print(f"Loaded {len(global_names)} existing names")
|
||||
|
||||
# Build available replacements (not already in corpus)
|
||||
available = [n for n in REPLACEMENT_POOL if n.lower() not in global_names]
|
||||
print(f"Available replacements: {len(available)} (from pool of {len(REPLACEMENT_POOL)})")
|
||||
|
||||
# Find all features that need replacement
|
||||
replacements_needed = []
|
||||
for markers_path in sorted(WIKI_SYSTEMS.glob("*/bodies/*/markers.json")):
|
||||
body_id = markers_path.parent.name
|
||||
m = json.loads(markers_path.read_text())
|
||||
for section in ("cities", "rivers", "oceans", "mountain_ranges", "pois"):
|
||||
for feat in m.get(section, []):
|
||||
name = feat.get("name", "")
|
||||
if name and name.lower() in FEWSHOT_NAMES:
|
||||
replacements_needed.append((markers_path, body_id, section, feat))
|
||||
|
||||
print(f"Features to replace: {len(replacements_needed)}")
|
||||
|
||||
if len(available) < len(replacements_needed):
|
||||
print(f"WARNING: only {len(available)} replacements for {len(replacements_needed)} features")
|
||||
print(" some features will keep their few-shot names")
|
||||
|
||||
# Assign replacements deterministically — hash body_id + feature_id
|
||||
# to pick from the pool, ensuring each body gets different names
|
||||
used_per_body = defaultdict(set)
|
||||
replacement_idx = 0
|
||||
changed_files = set()
|
||||
total_replaced = 0
|
||||
|
||||
for markers_path, body_id, section, feat in replacements_needed:
|
||||
old_name = feat["name"]
|
||||
|
||||
# Find next available name not yet used on this body
|
||||
assigned = None
|
||||
for attempt in range(len(available)):
|
||||
candidate = available[(replacement_idx + attempt) % len(available)]
|
||||
if candidate.lower() not in used_per_body[body_id]:
|
||||
assigned = candidate
|
||||
replacement_idx = (replacement_idx + attempt + 1) % len(available)
|
||||
break
|
||||
|
||||
if assigned is None:
|
||||
print(f" SKIP {body_id}/{section}: no unique replacement for \"{old_name}\"")
|
||||
continue
|
||||
|
||||
feat["name"] = assigned
|
||||
used_per_body[body_id].add(assigned.lower())
|
||||
global_names.add(assigned.lower())
|
||||
changed_files.add(markers_path)
|
||||
total_replaced += 1
|
||||
|
||||
# Write changed files
|
||||
for markers_path in changed_files:
|
||||
body_id = markers_path.parent.name
|
||||
m = json.loads(markers_path.read_text())
|
||||
|
||||
# Re-apply changes (re-read since we modified feat objects in memory)
|
||||
# Actually the feat dicts are still referenced — just rewrite
|
||||
# But we need to reload and re-match since we didn't track which file
|
||||
# has which changes...
|
||||
|
||||
# Simpler approach: reload, replace, write
|
||||
# Reset and do it properly
|
||||
replacement_idx = 0
|
||||
used_per_body = defaultdict(set)
|
||||
changed_bodies = []
|
||||
|
||||
# Group by file
|
||||
by_file = defaultdict(list)
|
||||
for markers_path, body_id, section, feat in replacements_needed:
|
||||
by_file[markers_path].append((body_id, section, feat["id"] if "id" in feat else None))
|
||||
|
||||
for markers_path, entries in by_file.items():
|
||||
body_id = markers_path.parent.name
|
||||
m = json.loads(markers_path.read_text())
|
||||
changed = False
|
||||
|
||||
for _, section, feat_id in entries:
|
||||
for feat in m.get(section, []):
|
||||
name = feat.get("name") or ""
|
||||
if not name or name.lower() not in FEWSHOT_NAMES:
|
||||
continue
|
||||
|
||||
assigned = None
|
||||
for attempt in range(len(available)):
|
||||
candidate = available[(replacement_idx + attempt) % len(available)]
|
||||
if candidate.lower() not in used_per_body[body_id]:
|
||||
assigned = candidate
|
||||
replacement_idx = (replacement_idx + attempt + 1) % len(available)
|
||||
break
|
||||
|
||||
if assigned:
|
||||
feat["name"] = assigned
|
||||
used_per_body[body_id].add(assigned.lower())
|
||||
changed = True
|
||||
|
||||
if changed:
|
||||
markers_path.write_text(json.dumps(m, indent=2) + "\n")
|
||||
changed_bodies.append(body_id)
|
||||
# Sync to DB if body exists
|
||||
try:
|
||||
sync_markers_to_db(conn, body_id, m)
|
||||
except Exception:
|
||||
pass # orphan body
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
print(f"\nReplaced few-shot names on {len(changed_bodies)} bodies")
|
||||
print("Done.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+370
-123
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
gemma_naming.py — Batch-name every empty name field in the reach's
|
||||
markers.json files using the Gemma 2 voice pipeline (#833, D-191 §4).
|
||||
markers.json files using the Gemma 4 E2B tooling pipeline (#833, D-191 §4).
|
||||
|
||||
Pipeline per body:
|
||||
1. Load markers.json; identify feature records whose `name` is empty
|
||||
@@ -40,7 +40,9 @@ import argparse
|
||||
import datetime
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
@@ -60,6 +62,10 @@ from generate_atlas import ( # noqa: E402
|
||||
)
|
||||
|
||||
import sqlite3 # noqa: E402
|
||||
from naming_core import ( # noqa: E402
|
||||
name_features_batch,
|
||||
mood_for_body,
|
||||
)
|
||||
|
||||
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
|
||||
WIKI_SYSTEMS = REPO_ROOT / "wiki" / "star-systems"
|
||||
@@ -91,6 +97,7 @@ _CAPTURE_FILE = None # set in main() when --dump-prompts is used
|
||||
# across worktrees (too large to duplicate).
|
||||
HOME_PROJECTS = Path.home() / "Projects" / "settled-reach"
|
||||
BINARIES_DIR = HOME_PROJECTS / "binaries"
|
||||
MODELS_DIR = HOME_PROJECTS / "models"
|
||||
MAIN_WORKDIR = Path("/var/mnt/data/projects/settled-reach/main")
|
||||
|
||||
|
||||
@@ -98,19 +105,31 @@ def _find_sr_voice() -> Path:
|
||||
"""Resolve the default sr-voice binary path.
|
||||
|
||||
Preference order:
|
||||
1. $HOME/Projects/settled-reach/binaries/sr-voice-rocm — persistent
|
||||
across worktree lifetimes, the canonical dev location.
|
||||
2. main workdir's target/release/sr-voice — legacy, for
|
||||
backward-compatibility with older layouts.
|
||||
1. $HOME/Projects/settled-reach/binaries/sr-voice-tooling — Gemma 4
|
||||
tooling binary, preferred for content generation.
|
||||
2. $HOME/Projects/settled-reach/binaries/sr-voice-rocm — Gemma 2
|
||||
ROCm binary, fallback.
|
||||
3. main workdir's target/release/sr-voice — legacy.
|
||||
"""
|
||||
tooling_bin = BINARIES_DIR / "sr-voice-tooling"
|
||||
if tooling_bin.exists():
|
||||
return tooling_bin
|
||||
rocm_bin = BINARIES_DIR / "sr-voice-rocm"
|
||||
if rocm_bin.exists():
|
||||
return rocm_bin
|
||||
return MAIN_WORKDIR / "server" / "sr-voice" / "target" / "release" / "sr-voice"
|
||||
|
||||
|
||||
def _find_default_model() -> Path:
|
||||
"""Resolve the default model path. Prefers Gemma 4 over Gemma 2."""
|
||||
gemma4 = MODELS_DIR / "gemma-4.gguf"
|
||||
if gemma4.exists():
|
||||
return gemma4
|
||||
return MAIN_WORKDIR / "server" / "models" / "gemma2.gguf"
|
||||
|
||||
|
||||
DEFAULT_SR_VOICE = _find_sr_voice()
|
||||
DEFAULT_MODEL = MAIN_WORKDIR / "server" / "models" / "gemma2.gguf"
|
||||
DEFAULT_MODEL = _find_default_model()
|
||||
MOCK_STDIO = REPO_ROOT / "server" / "sr-voice" / "mock-stdio.sh"
|
||||
|
||||
|
||||
@@ -285,12 +304,181 @@ def palette_for(corridor: str | None, system_id: str = "") -> dict[str, str]:
|
||||
All bodies in the same system get the same sub-style (consistent
|
||||
cultural register per star system). Different systems rotate through
|
||||
the sub-style list via hash(system_id).
|
||||
|
||||
This is the FALLBACK path — the preferred path is select_register()
|
||||
which asks Gemma to pick the register based on wiki/GTTR content.
|
||||
"""
|
||||
substyles = CORRIDOR_SUBSTYLES.get(corridor or "core", DEFAULT_SUBSTYLES)
|
||||
idx = int(hashlib.sha256(system_id.encode()).hexdigest()[:8], 16) % len(substyles)
|
||||
return substyles[idx]
|
||||
|
||||
|
||||
def _system_slug(system_id: str) -> str:
|
||||
"""Convert system_id ('GJ 411') to wiki directory slug ('GJ-411')."""
|
||||
if system_id.startswith("GJ "):
|
||||
return "GJ-" + system_id[3:]
|
||||
return system_id
|
||||
|
||||
|
||||
def load_wiki_context(system_id: str) -> tuple[str | None, str | None]:
|
||||
"""Read index.md and gttr.md for a system from wiki/star-systems/.
|
||||
|
||||
Returns (index_text, gttr_text). Either or both may be None if the
|
||||
file doesn't exist.
|
||||
"""
|
||||
slug = _system_slug(system_id)
|
||||
sys_dir = WIKI_SYSTEMS / slug
|
||||
index_path = sys_dir / "index.md"
|
||||
gttr_path = sys_dir / "gttr.md"
|
||||
index_text = index_path.read_text() if index_path.exists() else None
|
||||
gttr_text = gttr_path.read_text() if gttr_path.exists() else None
|
||||
return index_text, gttr_text
|
||||
|
||||
|
||||
def _extract_cultural_lines(wiki_text: str, max_lines: int = 8) -> str:
|
||||
"""Pull the most culturally relevant lines from a wiki index.md.
|
||||
|
||||
Scans for lines mentioning heritage, founding identity, language,
|
||||
cultural texture, or corridor affiliation. Falls back to the first
|
||||
prose paragraphs if no keyword hits. Keeps the excerpt short enough
|
||||
for Gemma 2 2B's 1024-token context.
|
||||
"""
|
||||
keywords = (
|
||||
"cultural", "heritage", "founding", "settler", "surname",
|
||||
"language", "tradition", "diaspora", "population carried",
|
||||
"portuguese", "iberian", "japanese", "korean", "chinese",
|
||||
"filipino", "german", "dutch", "nordic", "scandinavian",
|
||||
"polish", "czech", "finnish", "baltic", "swahili", "african",
|
||||
"angolan", "cape verde", "irish", "scottish", "australian",
|
||||
"british", "brazilian", "mozambic", "norwegian", "frisian",
|
||||
"afrikaans", "lusophone", "corridor",
|
||||
)
|
||||
hits: list[str] = []
|
||||
prose: list[str] = []
|
||||
for line in wiki_text.splitlines():
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped.startswith("#") or stripped.startswith("|") or stripped.startswith("---") or stripped.startswith("<!--"):
|
||||
continue
|
||||
low = stripped.lower()
|
||||
if any(kw in low for kw in keywords):
|
||||
hits.append(stripped[:200])
|
||||
elif len(prose) < max_lines:
|
||||
prose.append(stripped[:200])
|
||||
|
||||
selected = hits[:max_lines] if hits else prose[:max_lines]
|
||||
return "\n".join(selected)
|
||||
|
||||
|
||||
def select_register(
|
||||
voice: "VoiceSubprocess",
|
||||
corridor: str,
|
||||
system_id: str,
|
||||
wiki_text: str | None,
|
||||
gttr_text: str | None,
|
||||
log: "Logger",
|
||||
max_attempts: int = 3,
|
||||
) -> dict[str, str] | None:
|
||||
"""Ask Gemma to pick the best cultural register for this system.
|
||||
|
||||
Presents the corridor's sub-style options numbered 1..N alongside
|
||||
a compact cultural excerpt (gttr_hook + key wiki lines). Gemma
|
||||
replies with just the number. Returns the selected sub-style dict,
|
||||
or None if all attempts fail (caller falls back to hash-based
|
||||
palette_for).
|
||||
"""
|
||||
substyles = CORRIDOR_SUBSTYLES.get(corridor, DEFAULT_SUBSTYLES)
|
||||
if not wiki_text and not gttr_text:
|
||||
return None
|
||||
|
||||
# Build numbered option list — inflection only, no examples, to
|
||||
# save tokens. Gemma needs to match cultural identity, not mimic
|
||||
# example names.
|
||||
options: list[str] = []
|
||||
for idx, style in enumerate(substyles, 1):
|
||||
options.append(f"{idx}. {style['inflection']}")
|
||||
option_block = "\n".join(options)
|
||||
|
||||
# Build a compact context block that fits in ~400 tokens.
|
||||
# The gttr_hook is a 30-45 word summary; the wiki excerpt adds
|
||||
# the strongest cultural-identity lines.
|
||||
context_parts: list[str] = []
|
||||
if gttr_text:
|
||||
# Use the first substantive GTTR paragraph — skip the title
|
||||
# line (# THE DRIFTER'S GUIDE ...) and any blank lines.
|
||||
for para in gttr_text.strip().split("\n\n"):
|
||||
cleaned = para.replace("#", "").strip()
|
||||
# Skip title lines and section headers
|
||||
if cleaned.startswith("THE DRIFTER") or cleaned.startswith("DRIFTER"):
|
||||
continue
|
||||
if not cleaned or len(cleaned) < 20:
|
||||
continue
|
||||
context_parts.append(cleaned[:300])
|
||||
break
|
||||
if wiki_text:
|
||||
cultural = _extract_cultural_lines(wiki_text)
|
||||
if cultural:
|
||||
context_parts.append(cultural)
|
||||
|
||||
if not context_parts:
|
||||
return None
|
||||
|
||||
context = "\n".join(context_parts)
|
||||
|
||||
# Few-shot format: Gemma 2 2B is much better at pattern completion
|
||||
# than instruction following. Show 2 worked examples, then the
|
||||
# target system. Keep examples short and from different corridors
|
||||
# than the target so they don't bias the answer.
|
||||
# Fixed preamble + tail that frame the completion pattern.
|
||||
preamble = (
|
||||
"Match the star system to the best cultural naming register.\n\n"
|
||||
"System: Neustadt — German-heritage industrial town, west corridor, orderly municipal governance.\n"
|
||||
"1. German settlement 2. Dutch colonial 3. Nordic 4. Polish/Czech 5. Baltic/Finnish\n"
|
||||
"Best: 1\n\n"
|
||||
"System: Matsue — Japanese precision manufacturing hub, east corridor.\n"
|
||||
"1. Korean 2. Japanese 3. Taiwanese/Hakka 4. Filipino 5. Mixed East Asian\n"
|
||||
"Best: 2\n\n"
|
||||
"System: "
|
||||
)
|
||||
tail = f"\n{option_block}\nBest (number only):"
|
||||
|
||||
# Reserve tokens for preamble, tail, and a few output tokens.
|
||||
# Rough estimate: 1 token ≈ 4 chars for English prose.
|
||||
max_prompt_chars = (voice.ctx_size - 16) * 4 # 16 tokens headroom for output
|
||||
budget = max_prompt_chars - len(preamble) - len(tail)
|
||||
if budget < 100:
|
||||
budget = 100
|
||||
if len(context) > budget:
|
||||
context = context[:budget]
|
||||
|
||||
prompt = f"{preamble}{context}{tail}"
|
||||
|
||||
for attempt in range(max_attempts):
|
||||
seed = int(
|
||||
hashlib.sha256(
|
||||
f"register|{system_id}|{attempt}".encode()
|
||||
).hexdigest()[:8],
|
||||
16,
|
||||
)
|
||||
try:
|
||||
raw = voice.request(prompt, seed)
|
||||
except RuntimeError:
|
||||
continue
|
||||
|
||||
# Extract the first number from the response. Gemma may reply
|
||||
# "1", "1.", "Option 1", "Answer: 1", etc.
|
||||
digits = re.search(r"\d+", raw.strip() or "")
|
||||
if not digits:
|
||||
continue
|
||||
try:
|
||||
choice = int(digits.group())
|
||||
except ValueError:
|
||||
continue
|
||||
if 1 <= choice <= len(substyles):
|
||||
return substyles[choice - 1]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Feature prompt templates — few-shot format with rotating example pools
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -812,6 +1000,8 @@ def _build_prompt(
|
||||
f"or after places back home. Most names are mundane, short, and "
|
||||
f"direct — a surname, a compass direction, a feature, a practical "
|
||||
f"description. Classical or epic names are rare. "
|
||||
f"Avoid the obvious choice. Do not repeat the system or planet name. "
|
||||
f"Each name should be unique and surprising within its register. "
|
||||
f"Reply with ONLY the name, {cfg['length_hint']}, no brackets, "
|
||||
f"no quotes, no markdown, no label."
|
||||
)
|
||||
@@ -1156,6 +1346,7 @@ class VoiceSubprocess:
|
||||
refresh_every: int,
|
||||
verbose: bool,
|
||||
distrobox: str | None = None,
|
||||
ctx_size: int = 1024,
|
||||
):
|
||||
self.sr_voice_bin = sr_voice_bin
|
||||
self.model_path = model_path
|
||||
@@ -1163,6 +1354,7 @@ class VoiceSubprocess:
|
||||
self.refresh_every = max(1, refresh_every)
|
||||
self.verbose = verbose
|
||||
self.distrobox = distrobox
|
||||
self.ctx_size = ctx_size
|
||||
self.proc: subprocess.Popen | None = None
|
||||
self.request_count = 0
|
||||
self._start()
|
||||
@@ -1170,7 +1362,15 @@ class VoiceSubprocess:
|
||||
def _build_cmd(self) -> list[str]:
|
||||
if self.mock:
|
||||
return [str(MOCK_STDIO), "serve", "--stdio"]
|
||||
inner_cmd = [str(self.sr_voice_bin), "serve", "--stdio"]
|
||||
# sr-voice-tooling (Gemma 4) takes --model directly;
|
||||
# sr-voice (Gemma 2) needs `serve --stdio` subcommand.
|
||||
bin_name = self.sr_voice_bin.name if hasattr(self.sr_voice_bin, 'name') else str(self.sr_voice_bin).rsplit("/", 1)[-1]
|
||||
if "tooling" in bin_name:
|
||||
inner_cmd = [str(self.sr_voice_bin),
|
||||
"--ctx-size", str(self.ctx_size)]
|
||||
else:
|
||||
inner_cmd = [str(self.sr_voice_bin), "serve", "--stdio",
|
||||
"--ctx-size", str(self.ctx_size)]
|
||||
if self.model_path is not None:
|
||||
inner_cmd += ["--model", str(self.model_path)]
|
||||
# If a distrobox container was requested, invoke the binary
|
||||
@@ -1195,6 +1395,7 @@ class VoiceSubprocess:
|
||||
stderr=subprocess.DEVNULL if not self.verbose else None,
|
||||
text=True,
|
||||
bufsize=1, # line-buffered
|
||||
start_new_session=True, # own process group so _stop can kill the whole chain
|
||||
)
|
||||
self.request_count = 0
|
||||
# Mock prints a stderr banner synchronously; real sr-voice prints
|
||||
@@ -1255,12 +1456,32 @@ class VoiceSubprocess:
|
||||
self.proc.stdin.close()
|
||||
except Exception:
|
||||
pass
|
||||
# Kill the entire process group (distrobox → podman → sr-voice)
|
||||
# rather than just the top-level shell. Without this, the inner
|
||||
# sr-voice binary survives terminate() and holds the GPU.
|
||||
pgid = None
|
||||
try:
|
||||
self.proc.terminate()
|
||||
pgid = os.getpgid(self.proc.pid)
|
||||
except (ProcessLookupError, OSError):
|
||||
pass
|
||||
try:
|
||||
if pgid:
|
||||
os.killpg(pgid, signal.SIGTERM)
|
||||
else:
|
||||
self.proc.terminate()
|
||||
self.proc.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.proc.kill()
|
||||
self.proc.wait()
|
||||
try:
|
||||
if pgid:
|
||||
os.killpg(pgid, signal.SIGKILL)
|
||||
else:
|
||||
self.proc.kill()
|
||||
except (ProcessLookupError, OSError):
|
||||
pass
|
||||
try:
|
||||
self.proc.wait(timeout=3)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
self.proc = None
|
||||
@@ -1319,6 +1540,7 @@ def name_feature(
|
||||
log: "Logger",
|
||||
verbose: bool,
|
||||
max_attempts: int = 5,
|
||||
palette_override: dict[str, str] | None = None,
|
||||
) -> str:
|
||||
"""Request a name from Gemma, enforce blocklist + per-system dedup +
|
||||
per-body cross-type dedup, skip on persistent failure.
|
||||
@@ -1329,12 +1551,16 @@ def name_feature(
|
||||
- `body_used` — per-body set across ALL feature types. Prevents
|
||||
the same name from appearing as a river AND an ocean AND a
|
||||
mountain range on the same world.
|
||||
|
||||
When `palette_override` is set, it replaces the hash-based palette
|
||||
selection — used when select_register() picked a wiki-grounded
|
||||
cultural register for this system.
|
||||
"""
|
||||
corridor = ctx.get("cultural_corridor") or "core"
|
||||
if feature_type not in _PROMPT_CONFIG:
|
||||
return fallback_name(corridor, feature_type, _seed_for(world_seed, body_id, local_id, 0))
|
||||
|
||||
palette = palette_for(corridor, system_id)
|
||||
palette = palette_override or palette_for(corridor, system_id)
|
||||
planet_class = ctx.get("planet_class") or "habitable"
|
||||
|
||||
dedup_key = (system_id, feature_type)
|
||||
@@ -1489,23 +1715,38 @@ def load_system_gttr_hooks(conn: sqlite3.Connection) -> dict[str, str]:
|
||||
return {row[0]: row[1] for row in rows}
|
||||
|
||||
|
||||
def load_body_hop_order(conn: sqlite3.Connection) -> dict[str, tuple[int, str]]:
|
||||
"""Return a `{body_id: (hop_distance_from_gateway, body_id)}` map used
|
||||
as a stable sort key so the pipeline walks the reach from core
|
||||
outward: Gateway (hop 0) first, then hop 1, hop 2, ... all the way
|
||||
to the deep frontier. Ordering core-first gives those bodies first
|
||||
crack at every unique Gemma output and lets outer sectors fall
|
||||
into the palette fallback when they lose the dedup race.
|
||||
def load_body_hop_order(conn: sqlite3.Connection) -> dict[str, tuple[int, str, int, int, str]]:
|
||||
"""Return a `{body_id: (hop, system_id, uninhabited, neg_pop, body_id)}`
|
||||
sort-key map. The pipeline walks the reach from core outward, keeps
|
||||
all bodies in the same system together, and within each system
|
||||
processes inhabited bodies first (sorted by population descending).
|
||||
This gives the most important worlds — the ones players will
|
||||
actually visit — first pick of the cultural register's vocabulary.
|
||||
Barren moons get whatever's left or the adjacent-register refill,
|
||||
which is fine for star-map dressing.
|
||||
|
||||
Sort key components:
|
||||
- hop: system distance from Gateway (0 = core, higher = frontier)
|
||||
- system_id: groups all bodies in same system together
|
||||
- uninhabited: 0 for inhabited, 1 for uninhabited (inhabited first)
|
||||
- neg_pop: negative population (higher pop sorts first)
|
||||
- body_id: tiebreaker for determinism
|
||||
"""
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT b.body_id,
|
||||
COALESCE(sg.hop_distance_from_gateway, 99) AS hop
|
||||
b.system_id,
|
||||
COALESCE(sg.hop_distance_from_gateway, 99) AS hop,
|
||||
b.inhabited,
|
||||
COALESCE(b.population, 0) AS pop
|
||||
FROM bodies b
|
||||
LEFT JOIN system_gates sg ON b.system_id = sg.system_id
|
||||
"""
|
||||
).fetchall()
|
||||
return {body_id: (hop, body_id) for body_id, hop in rows}
|
||||
return {
|
||||
body_id: (hop, system_id, 0 if inhabited else 1, -(pop or 0), body_id)
|
||||
for body_id, system_id, hop, inhabited, pop in rows
|
||||
}
|
||||
|
||||
|
||||
def _is_blank(value) -> bool:
|
||||
@@ -1549,6 +1790,7 @@ def process_body(
|
||||
hop: int,
|
||||
log: "Logger",
|
||||
verbose: bool,
|
||||
palette_override: dict[str, str] | None = None,
|
||||
) -> dict:
|
||||
"""Fill every empty name field in this body's markers.json. Returns
|
||||
a summary counts dict plus a `generated` dict mapping section name
|
||||
@@ -1588,106 +1830,80 @@ def process_body(
|
||||
if name and isinstance(name, str) and name.strip():
|
||||
body_used.add(name.strip())
|
||||
|
||||
# Cities
|
||||
for city in markers.get("cities") or []:
|
||||
if not _is_blank(city.get("name")):
|
||||
corpus.setdefault((system_id, _feature_type_for_city(city)), set()).add(
|
||||
city["name"]
|
||||
)
|
||||
counts["preserved"] += 1
|
||||
continue
|
||||
feature_type = _feature_type_for_city(city)
|
||||
name = name_feature(
|
||||
voice, feature_type, ctx, blocklist, corpus, body_used,
|
||||
system_hook, system_id,
|
||||
world_seed, body_id, city.get("id") or "city_?",
|
||||
hop, log, verbose,
|
||||
)
|
||||
if name is not None:
|
||||
city["name"] = name
|
||||
counts["cities"] += 1
|
||||
generated["cities"].append(name)
|
||||
changed = True
|
||||
# Batch naming: group blank features by type, request N names at
|
||||
# once, rank by Levenshtein distinctiveness, assign.
|
||||
inflection = (palette_override or palette_for(corridor, system_id))["inflection"]
|
||||
substyles = CORRIDOR_SUBSTYLES.get(corridor, DEFAULT_SUBSTYLES)
|
||||
mood = mood_for_body(body_id, world_seed)
|
||||
|
||||
# Rivers
|
||||
for river in markers.get("rivers") or []:
|
||||
if not _is_blank(river.get("name")):
|
||||
corpus.setdefault((system_id, "river"), set()).add(river["name"])
|
||||
counts["preserved"] += 1
|
||||
continue
|
||||
name = name_feature(
|
||||
voice, "river", ctx, blocklist, corpus, body_used,
|
||||
system_hook, system_id,
|
||||
world_seed, body_id, river.get("id") or "river_?",
|
||||
hop, log, verbose,
|
||||
)
|
||||
if name is not None:
|
||||
river["name"] = name
|
||||
counts["rivers"] += 1
|
||||
generated["rivers"].append(name)
|
||||
changed = True
|
||||
# Helper: batch-name blank features in a marker section
|
||||
def _batch_fill(
|
||||
section_key: str,
|
||||
feature_type_fn, # callable(feat) -> str
|
||||
count_key: str,
|
||||
):
|
||||
nonlocal changed
|
||||
features = markers.get(section_key) or []
|
||||
# Separate preserved vs blank
|
||||
blank = []
|
||||
for feat in features:
|
||||
if not _is_blank(feat.get("name")):
|
||||
ft = feature_type_fn(feat)
|
||||
corpus.setdefault((system_id, ft), set()).add(feat["name"])
|
||||
counts["preserved"] += 1
|
||||
else:
|
||||
blank.append(feat)
|
||||
|
||||
# Oceans / seas / lakes
|
||||
for water in markers.get("oceans") or []:
|
||||
if not _is_blank(water.get("name")):
|
||||
corpus.setdefault((system_id, _feature_type_for_ocean(water)), set()).add(
|
||||
water["name"]
|
||||
)
|
||||
counts["preserved"] += 1
|
||||
continue
|
||||
feature_type = _feature_type_for_ocean(water)
|
||||
name = name_feature(
|
||||
voice, feature_type, ctx, blocklist, corpus, body_used,
|
||||
system_hook, system_id,
|
||||
world_seed, body_id, water.get("id") or "water_?",
|
||||
hop, log, verbose,
|
||||
)
|
||||
if name is not None:
|
||||
water["name"] = name
|
||||
counts["oceans"] += 1
|
||||
generated["oceans"].append(name)
|
||||
changed = True
|
||||
if not blank:
|
||||
return
|
||||
|
||||
# Mountain ranges
|
||||
for rng_feat in markers.get("mountain_ranges") or []:
|
||||
if not _is_blank(rng_feat.get("name")):
|
||||
corpus.setdefault((system_id, "mountain_range"), set()).add(
|
||||
rng_feat["name"]
|
||||
)
|
||||
counts["preserved"] += 1
|
||||
continue
|
||||
name = name_feature(
|
||||
voice, "mountain_range", ctx, blocklist, corpus, body_used,
|
||||
system_hook, system_id,
|
||||
world_seed, body_id, rng_feat.get("id") or "range_?",
|
||||
hop, log, verbose,
|
||||
)
|
||||
if name is not None:
|
||||
rng_feat["name"] = name
|
||||
counts["mountain_ranges"] += 1
|
||||
generated["mountain_ranges"].append(name)
|
||||
changed = True
|
||||
# Group blanks by feature_type (cities have capital/secondary,
|
||||
# pois have transit/institutional/cultural, oceans have ocean/sea/lake)
|
||||
by_type: dict[str, list[dict]] = {}
|
||||
for feat in blank:
|
||||
ft = feature_type_fn(feat)
|
||||
by_type.setdefault(ft, []).append(feat)
|
||||
|
||||
# POIs
|
||||
for poi in markers.get("pois") or []:
|
||||
if not _is_blank(poi.get("name")):
|
||||
corpus.setdefault((system_id, _feature_type_for_poi(poi)), set()).add(
|
||||
poi["name"]
|
||||
for ft, feats in by_type.items():
|
||||
need = len(feats)
|
||||
# Build taken list from corpus (cross-body dedup)
|
||||
taken = list(corpus.get((system_id, ft), set()))
|
||||
# Also include body_used to avoid cross-type collisions on same body
|
||||
taken_full = taken + list(body_used)
|
||||
|
||||
names = name_features_batch(
|
||||
voice=voice,
|
||||
feature_type=ft,
|
||||
count=need,
|
||||
inflection=inflection,
|
||||
corridor=corridor,
|
||||
corridor_substyles=substyles,
|
||||
taken=taken_full,
|
||||
prompt_config=_PROMPT_CONFIG,
|
||||
system_name=ctx.get("system_proper_name"),
|
||||
body_name=ctx.get("body_proper_name"),
|
||||
system_hook=system_hook,
|
||||
mood=mood,
|
||||
body_id=body_id,
|
||||
world_seed=world_seed,
|
||||
ctx_size=voice.ctx_size,
|
||||
)
|
||||
counts["preserved"] += 1
|
||||
continue
|
||||
feature_type = _feature_type_for_poi(poi)
|
||||
name = name_feature(
|
||||
voice, feature_type, ctx, blocklist, corpus, body_used,
|
||||
system_hook, system_id,
|
||||
world_seed, body_id, poi.get("id") or "poi_?",
|
||||
hop, log, verbose,
|
||||
)
|
||||
if name is not None:
|
||||
poi["name"] = name
|
||||
counts["pois"] += 1
|
||||
generated["pois"].append(name)
|
||||
changed = True
|
||||
|
||||
# Assign names to features in order
|
||||
for i, feat in enumerate(feats):
|
||||
if i < len(names):
|
||||
feat["name"] = names[i]
|
||||
counts[count_key] += 1
|
||||
generated[count_key].append(names[i])
|
||||
corpus.setdefault((system_id, ft), set()).add(names[i])
|
||||
body_used.add(names[i])
|
||||
changed = True
|
||||
|
||||
_batch_fill("cities", _feature_type_for_city, "cities")
|
||||
_batch_fill("rivers", lambda f: "river", "rivers")
|
||||
_batch_fill("oceans", _feature_type_for_ocean, "oceans")
|
||||
_batch_fill("mountain_ranges", lambda f: "mountain_range", "mountain_ranges")
|
||||
_batch_fill("pois", _feature_type_for_poi, "pois")
|
||||
|
||||
if changed:
|
||||
markers_path.write_text(json.dumps(markers, indent=2) + "\n")
|
||||
@@ -1696,8 +1912,14 @@ def process_body(
|
||||
# immediately so a mid-run crash / kill loses at most one body
|
||||
# of DB state — the markers.json files are already persisted
|
||||
# above, atomically, via Path.write_text.
|
||||
sync_markers_to_db(conn, body_id, markers)
|
||||
conn.commit()
|
||||
# Skip DB sync for orphan bodies (markers.json exists but no
|
||||
# row in the bodies table — FK constraint would fail).
|
||||
try:
|
||||
sync_markers_to_db(conn, body_id, markers)
|
||||
conn.commit()
|
||||
except Exception as e:
|
||||
conn.rollback()
|
||||
log(f" DB sync skipped for {body_id}: {e}")
|
||||
|
||||
counts["generated"] = generated
|
||||
return counts
|
||||
@@ -1734,9 +1956,9 @@ def discover_bodies(
|
||||
p for p in all_markers if _body_id_from_path(p)[0] == body_filter
|
||||
]
|
||||
|
||||
def sort_key(path: Path) -> tuple[int, str]:
|
||||
body_id = _body_id_from_path(path)[0]
|
||||
return hop_order.get(body_id, (99, body_id))
|
||||
def sort_key(path: Path) -> tuple[int, str, int, int, str]:
|
||||
body_id, system_id = _body_id_from_path(path)
|
||||
return hop_order.get(body_id, (99, system_id, 1, 0, body_id))
|
||||
|
||||
all_markers.sort(key=sort_key)
|
||||
|
||||
@@ -1874,6 +2096,15 @@ def main():
|
||||
|
||||
hop_order = load_body_hop_order(conn)
|
||||
system_gttr_hooks = load_system_gttr_hooks(conn)
|
||||
# Preload corridor for each system so select_register can scope to
|
||||
# the right sub-style list without re-querying per body.
|
||||
system_corridors: dict[str, str] = {
|
||||
row[0]: row[1] or "core"
|
||||
for row in conn.execute(
|
||||
"SELECT system_id, COALESCE(cultural_corridor, geographic_sector, 'core') "
|
||||
"FROM star_systems"
|
||||
).fetchall()
|
||||
}
|
||||
markers_paths = discover_bodies(args.body, args.limit, hop_order)
|
||||
if not markers_paths:
|
||||
log(f"error: no markers.json found (body={args.body})")
|
||||
@@ -1926,6 +2157,7 @@ def main():
|
||||
).fetchall()
|
||||
}
|
||||
last_system_id: str | None = None
|
||||
current_palette: dict[str, str] | None = None
|
||||
|
||||
totals = {
|
||||
"cities": 0, "rivers": 0, "oceans": 0, "mountain_ranges": 0, "pois": 0,
|
||||
@@ -1953,16 +2185,30 @@ def main():
|
||||
if system_id != last_system_id:
|
||||
last_system_id = system_id
|
||||
sys_proper = system_name_cache.get(system_id, "")
|
||||
sys_hop = hop_order.get(body_id, (99, ""))[0]
|
||||
sys_hop = hop_order.get(body_id, (99, "", 1, 0, ""))[0]
|
||||
label = f"{system_id}"
|
||||
if sys_proper:
|
||||
label = f"{system_id} — {sys_proper}"
|
||||
log(f" ── SYSTEM {len(seen_systems)+1}/{total_systems} "
|
||||
f"{label} (hop {sys_hop})")
|
||||
|
||||
# Ask Gemma to pick the cultural register based on
|
||||
# wiki content instead of the hash-based randomizer.
|
||||
corridor = system_corridors.get(system_id, "core")
|
||||
wiki_index, wiki_gttr = load_wiki_context(system_id)
|
||||
current_palette = select_register(
|
||||
voice, corridor, system_id,
|
||||
wiki_index, wiki_gttr, log,
|
||||
)
|
||||
if current_palette:
|
||||
log(f" register: {current_palette['inflection']}")
|
||||
else:
|
||||
current_palette = palette_for(corridor, system_id)
|
||||
log(f" register: {current_palette['inflection']} (hash fallback)")
|
||||
|
||||
seen_systems.add(system_id)
|
||||
t0 = time.time()
|
||||
hop = hop_order.get(body_id, (99, ""))[0]
|
||||
hop = hop_order.get(body_id, (99, "", 1, 0, ""))[0]
|
||||
counts = process_body(
|
||||
body_id=body_id,
|
||||
system_id=system_id,
|
||||
@@ -1976,12 +2222,13 @@ def main():
|
||||
hop=hop,
|
||||
log=log,
|
||||
verbose=args.verbose,
|
||||
palette_override=current_palette,
|
||||
)
|
||||
elapsed = time.time() - t0
|
||||
|
||||
body_progress = f"body {i+1}/{len(markers_paths)}"
|
||||
sys_progress = f"sys {len(seen_systems)}/{total_systems}"
|
||||
hop = hop_order.get(body_id, (99, ""))[0]
|
||||
hop = hop_order.get(body_id, (99, "", 1, 0, ""))[0]
|
||||
progress = f"{body_progress} {sys_progress} hop={hop}"
|
||||
|
||||
# Append the body's proper name if it has one ("Threshold",
|
||||
|
||||
@@ -0,0 +1,395 @@
|
||||
"""naming_core.py — Shared algorithms for the atlas naming pipeline.
|
||||
|
||||
Contains: Levenshtein distance, distinctiveness ranking, batch prompt
|
||||
building, mood pool, name validation, and response parsing. Used by
|
||||
both gemma_naming.py (production pipeline) and test scripts.
|
||||
|
||||
Version history:
|
||||
0.1 2026-04-16 Initial extraction from test_batch_naming.py
|
||||
- levenshtein, word_avg_distance, select_distinct
|
||||
- build_batch_prompt with mood injection
|
||||
- parse_batch_response with dedup
|
||||
- name_features_batch with adjacent-register refill
|
||||
- is_valid_name prompt-fragment filter
|
||||
0.2 2026-04-17 Post-QA hardening
|
||||
- few-shot example blocklist (prevents prompt bleed)
|
||||
- minimum name length raised to 3 chars
|
||||
- bracket/number rejection in is_valid_name
|
||||
- parse_batch_response filters few-shot examples
|
||||
"""
|
||||
|
||||
__version__ = "0.2"
|
||||
|
||||
import hashlib
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mood pool — randomized per-body emotional seed for vocabulary divergence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
MOOD_POOL = [
|
||||
"ambition", "family", "wealth", "community", "industry",
|
||||
"pride", "fleeting", "hope", "fear", "isolation",
|
||||
"devotion", "defiance", "loss",
|
||||
]
|
||||
|
||||
|
||||
def mood_for_body(body_id: str, world_seed: int = 42) -> str:
|
||||
"""Deterministic mood selection per body."""
|
||||
h = int(hashlib.sha256(f"mood|{body_id}|{world_seed}".encode()).hexdigest()[:8], 16)
|
||||
return MOOD_POOL[h % len(MOOD_POOL)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Levenshtein distance and distinctiveness ranking
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def levenshtein(a: str, b: str) -> int:
|
||||
"""Standard Levenshtein edit distance."""
|
||||
if len(a) < len(b):
|
||||
return levenshtein(b, a)
|
||||
if not b:
|
||||
return len(a)
|
||||
prev = list(range(len(b) + 1))
|
||||
for i, ca in enumerate(a):
|
||||
curr = [i + 1]
|
||||
for j, cb in enumerate(b):
|
||||
cost = 0 if ca == cb else 1
|
||||
curr.append(min(curr[j] + 1, prev[j + 1] + 1, prev[j] + cost))
|
||||
prev = curr
|
||||
return prev[-1]
|
||||
|
||||
|
||||
def _words(name: str) -> list[str]:
|
||||
"""Split a name into lowercase words for per-word comparison."""
|
||||
return [w for w in name.lower().split() if w]
|
||||
|
||||
|
||||
def word_avg_distance(a: str, b: str) -> float:
|
||||
"""Average Levenshtein distance across word pairs.
|
||||
|
||||
Compares each word in the shorter name against the closest word in
|
||||
the longer name, then averages. Shared structural words (Serra,
|
||||
The, Mount) lower the score but don't block — the unique words
|
||||
pull the average up.
|
||||
|
||||
Returns 0.0 for identical, higher = more distinct.
|
||||
"""
|
||||
wa, wb = _words(a), _words(b)
|
||||
if not wa or not wb:
|
||||
return float(levenshtein(a.lower(), b.lower()))
|
||||
|
||||
shorter, longer = (wa, wb) if len(wa) <= len(wb) else (wb, wa)
|
||||
total = 0.0
|
||||
for sw in shorter:
|
||||
best = min(levenshtein(sw, lw) for lw in longer)
|
||||
total += best
|
||||
return total / len(shorter)
|
||||
|
||||
|
||||
def min_avg_distance_to_set(name: str, existing: list[str]) -> float:
|
||||
"""Minimum word-average distance from name to any name in the set."""
|
||||
if not existing:
|
||||
return 999.0
|
||||
return min(word_avg_distance(name, e) for e in existing)
|
||||
|
||||
|
||||
def select_distinct(
|
||||
candidates: list[str],
|
||||
count: int,
|
||||
taken: list[str],
|
||||
) -> list[str]:
|
||||
"""Greedily select the N most distinct names from candidates.
|
||||
|
||||
No hard rejection — all candidates are eligible except exact
|
||||
matches to taken names. Ranked by distinctiveness (word-average
|
||||
Levenshtein distance) against both taken names and previously
|
||||
selected names. The most distinct candidate is picked first, then
|
||||
the next most distinct relative to the growing set, until the
|
||||
quota is filled.
|
||||
|
||||
Shared words (Serra, The, Forum) lower a candidate's preference
|
||||
but never block it. The greedy approach naturally spaces out
|
||||
selections.
|
||||
"""
|
||||
# Hard-filter exact matches to taken (case-insensitive)
|
||||
taken_lower = {t.lower() for t in taken}
|
||||
pool = [c for c in candidates if c.lower() not in taken_lower]
|
||||
|
||||
selected: list[str] = []
|
||||
reference: list[str] = list(taken)
|
||||
|
||||
for _ in range(min(count, len(pool))):
|
||||
if not pool:
|
||||
break
|
||||
|
||||
best_idx = 0
|
||||
best_score = -1.0
|
||||
for i, name in enumerate(pool):
|
||||
score = min_avg_distance_to_set(name, reference) if reference else 999.0
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_idx = i
|
||||
|
||||
chosen = pool.pop(best_idx)
|
||||
selected.append(chosen)
|
||||
reference.append(chosen)
|
||||
|
||||
return selected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Name validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Few-shot examples used in batch prompts. These must be blocked from
|
||||
# appearing as output — Gemma pattern-completes them verbatim, and
|
||||
# without this blocklist they appear 50-100x across the reach.
|
||||
FEWSHOT_BLOCKLIST = {
|
||||
"glen moray", "dunvegan ridge", "torridon", "cairn brae",
|
||||
"the kelpie's spine", "kloosterbeek", "nieuw rijn",
|
||||
"hoogland run", "van diemen's creek",
|
||||
}
|
||||
|
||||
|
||||
def is_valid_name(name: str) -> bool:
|
||||
"""Filter out garbage: too short, too long, contains periods/brackets,
|
||||
looks like a prompt fragment, matches a few-shot example, or contains
|
||||
digits."""
|
||||
if not name or len(name) < 3 or len(name) > 50:
|
||||
return False
|
||||
# Brackets, periods, digits — structural garbage
|
||||
if "." in name or "(" in name or ")" in name or "[" in name or "]" in name:
|
||||
return False
|
||||
if any(c.isdigit() for c in name):
|
||||
return False
|
||||
low = name.lower()
|
||||
# Prompt fragment echoes
|
||||
reject_phrases = [
|
||||
"names", "style:", "answer:", "must be", "avoid", "distinct",
|
||||
"already used", "do not", "need", "generate", "list",
|
||||
"number only", "comma-separated", "best:",
|
||||
]
|
||||
if any(phrase in low for phrase in reject_phrases):
|
||||
return False
|
||||
# Few-shot example bleed
|
||||
if low in FEWSHOT_BLOCKLIST:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def parse_batch_response(raw: str) -> list[str]:
|
||||
"""Parse a batch naming response into a list of clean, unique name strings.
|
||||
|
||||
Takes the first line only (model often continues with explanations
|
||||
or more styles), splits on commas, strips quotes/whitespace,
|
||||
filters invalid names, and deduplicates (preserving order).
|
||||
"""
|
||||
first_line = raw.strip().split("\n")[0] if raw.strip() else ""
|
||||
candidates = [
|
||||
n.strip().strip('"').strip("'").strip()
|
||||
for n in first_line.split(",")
|
||||
]
|
||||
# Deduplicate preserving order (model often repeats names in batch)
|
||||
seen: set[str] = set()
|
||||
unique: list[str] = []
|
||||
for n in candidates:
|
||||
if is_valid_name(n) and n.lower() not in seen:
|
||||
unique.append(n)
|
||||
seen.add(n.lower())
|
||||
return unique
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Batch prompt building
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def build_batch_prompt(
|
||||
feature_type: str,
|
||||
inflection: str,
|
||||
count: int,
|
||||
taken: list[str],
|
||||
prompt_config: dict,
|
||||
system_name: str | None = None,
|
||||
body_name: str | None = None,
|
||||
system_hook: str | None = None,
|
||||
mood: str | None = None,
|
||||
ctx_size: int = 1024,
|
||||
) -> str:
|
||||
"""Build a batch naming prompt asking for N names in one call.
|
||||
|
||||
Uses the same preamble structure as the single-name prompts but
|
||||
with few-shot examples showing comma-separated lists. The model
|
||||
pattern-completes the list.
|
||||
|
||||
The prompt is truncated to fit within ctx_size tokens (rough
|
||||
estimate: 1 token ≈ 4 chars).
|
||||
"""
|
||||
cfg = prompt_config.get(feature_type)
|
||||
if not cfg:
|
||||
return ""
|
||||
|
||||
subject = cfg["subject"]
|
||||
verb = "called" if subject.startswith("their ") else "named"
|
||||
|
||||
mood_clause = ""
|
||||
if mood:
|
||||
mood_clause = f" The settlers here had a sense of {mood}. "
|
||||
|
||||
preamble = (
|
||||
f"Settlers {verb} {subject} after themselves, after what they saw, "
|
||||
f"or after places back home. Most names are mundane, short, and "
|
||||
f"direct — a surname, a compass direction, a feature, a practical "
|
||||
f"description.{mood_clause}Avoid the obvious choice. Each name must "
|
||||
f"be distinct — no two names may share a root word.\n"
|
||||
f"Reply with ONLY a comma-separated list, no numbering, no markdown."
|
||||
)
|
||||
|
||||
lines = [preamble, ""]
|
||||
if system_name or body_name:
|
||||
ident = []
|
||||
if system_name:
|
||||
ident.append(f"System: {system_name}")
|
||||
if body_name:
|
||||
ident.append(f"Planet: {body_name}")
|
||||
lines.append(". ".join(ident) + ".")
|
||||
if system_hook:
|
||||
lines.append(f"About the system: {system_hook}")
|
||||
if taken:
|
||||
lines.append(f"Already used (do NOT repeat): {', '.join(taken)}")
|
||||
if system_name or body_name or system_hook or taken:
|
||||
lines.append("")
|
||||
|
||||
# Few-shot examples showing batch format
|
||||
lines.append("Style: Scottish Highland. 5 names: Glen Moray, Dunvegan Ridge, Torridon, Cairn Brae, The Kelpie's Spine")
|
||||
lines.append("Style: Dutch colonial. 4 names: Kloosterbeek, Nieuw Rijn, Hoogland Run, Van Diemen's Creek")
|
||||
lines.append("")
|
||||
|
||||
ask_for = count * 2 # oversample, then rank by distinctiveness
|
||||
tail = f"Style: {inflection}. {ask_for} names:"
|
||||
|
||||
# Truncate context to fit within ctx_size
|
||||
fixed = "\n".join(lines)
|
||||
max_chars = (ctx_size - 16) * 4 # 16 tokens headroom for output
|
||||
budget = max_chars - len(fixed) - len(tail) - 2 # 2 for newlines
|
||||
if budget < 0:
|
||||
# Trim the taken list to fit
|
||||
while taken and budget < 0:
|
||||
taken = taken[:-1]
|
||||
lines_rebuild = [preamble, ""]
|
||||
if system_name or body_name:
|
||||
ident = []
|
||||
if system_name:
|
||||
ident.append(f"System: {system_name}")
|
||||
if body_name:
|
||||
ident.append(f"Planet: {body_name}")
|
||||
lines_rebuild.append(". ".join(ident) + ".")
|
||||
if system_hook:
|
||||
lines_rebuild.append(f"About the system: {system_hook}")
|
||||
if taken:
|
||||
lines_rebuild.append(f"Already used (do NOT repeat): {', '.join(taken)}")
|
||||
lines_rebuild.append("")
|
||||
lines_rebuild.append("Style: Scottish Highland. 5 names: Glen Moray, Dunvegan Ridge, Torridon, Cairn Brae, The Kelpie's Spine")
|
||||
lines_rebuild.append("Style: Dutch colonial. 4 names: Kloosterbeek, Nieuw Rijn, Hoogland Run, Van Diemen's Creek")
|
||||
lines_rebuild.append("")
|
||||
fixed = "\n".join(lines_rebuild)
|
||||
budget = max_chars - len(fixed) - len(tail) - 2
|
||||
|
||||
return fixed + "\n" + tail
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Batch naming with refill
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def name_features_batch(
|
||||
voice, # VoiceSubprocess — not typed to avoid circular import
|
||||
feature_type: str,
|
||||
count: int,
|
||||
inflection: str,
|
||||
corridor: str,
|
||||
corridor_substyles: list[dict[str, str]],
|
||||
taken: list[str],
|
||||
prompt_config: dict,
|
||||
system_name: str | None,
|
||||
body_name: str | None,
|
||||
system_hook: str | None,
|
||||
mood: str | None,
|
||||
body_id: str,
|
||||
world_seed: int,
|
||||
ctx_size: int = 1024,
|
||||
) -> list[str]:
|
||||
"""Generate `count` names for a feature type using batch prompting.
|
||||
|
||||
Flow:
|
||||
1. Build a batch prompt asking for count*2 names.
|
||||
2. Send to voice, parse response.
|
||||
3. Rank by distinctiveness via Levenshtein, pick top `count`.
|
||||
4. If short, refill from the next adjacent register in the corridor.
|
||||
5. Return the final list of names.
|
||||
"""
|
||||
prompt = build_batch_prompt(
|
||||
feature_type=feature_type,
|
||||
inflection=inflection,
|
||||
count=count,
|
||||
taken=taken,
|
||||
prompt_config=prompt_config,
|
||||
system_name=system_name,
|
||||
body_name=body_name,
|
||||
system_hook=system_hook,
|
||||
mood=mood,
|
||||
ctx_size=ctx_size,
|
||||
)
|
||||
|
||||
seed = int(hashlib.sha256(
|
||||
f"batch|{body_id}|{feature_type}|{world_seed}".encode()
|
||||
).hexdigest()[:8], 16)
|
||||
|
||||
try:
|
||||
raw = voice.request(prompt, seed)
|
||||
except RuntimeError:
|
||||
raw = ""
|
||||
|
||||
candidates = parse_batch_response(raw)
|
||||
selected = select_distinct(candidates, count, taken)
|
||||
|
||||
# Refill from adjacent register if we didn't fill the quota
|
||||
if len(selected) < count and corridor_substyles:
|
||||
shortfall = count - len(selected)
|
||||
refill_taken = taken + selected
|
||||
|
||||
# Find the primary register's index and pick the next one
|
||||
primary_idx = next(
|
||||
(i for i, s in enumerate(corridor_substyles)
|
||||
if s["inflection"] == inflection),
|
||||
0,
|
||||
)
|
||||
refill_idx = (primary_idx + 1) % len(corridor_substyles)
|
||||
refill_inflection = corridor_substyles[refill_idx]["inflection"]
|
||||
|
||||
refill_prompt = build_batch_prompt(
|
||||
feature_type=feature_type,
|
||||
inflection=refill_inflection,
|
||||
count=shortfall * 3,
|
||||
taken=refill_taken,
|
||||
prompt_config=prompt_config,
|
||||
system_name=system_name,
|
||||
body_name=body_name,
|
||||
system_hook=system_hook,
|
||||
mood=mood,
|
||||
ctx_size=ctx_size,
|
||||
)
|
||||
|
||||
refill_seed = int(hashlib.sha256(
|
||||
f"refill|{body_id}|{feature_type}|{world_seed}".encode()
|
||||
).hexdigest()[:8], 16)
|
||||
|
||||
try:
|
||||
refill_raw = voice.request(refill_prompt, refill_seed)
|
||||
except RuntimeError:
|
||||
refill_raw = ""
|
||||
|
||||
refill_candidates = parse_batch_response(refill_raw)
|
||||
extra = select_distinct(refill_candidates, shortfall, refill_taken)
|
||||
selected.extend(extra)
|
||||
|
||||
return selected
|
||||
@@ -0,0 +1,523 @@
|
||||
#!/usr/bin/env python3
|
||||
"""QA report on atlas naming quality and distribution.
|
||||
|
||||
Runs checks against systems.db and markers.json files:
|
||||
- Exact duplicates within systems
|
||||
- Stem repetition (shared root words)
|
||||
- Prompt fragment leaks
|
||||
- Register bleed (wrong cultural register for corridor)
|
||||
- Feature-type mismatches (street names as mountains, etc.)
|
||||
- Short/long name outliers
|
||||
- Body-name echo (planet name used as feature stem)
|
||||
- Coverage gaps
|
||||
- Distribution by corridor and register
|
||||
|
||||
Usage:
|
||||
python3 tooling/planet-gen/qa_naming.py
|
||||
python3 tooling/planet-gen/qa_naming.py --verbose
|
||||
"""
|
||||
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
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"
|
||||
|
||||
sys.path.insert(0, str(TOOLING_DIR))
|
||||
from naming_core import _words
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def load_all_names(conn):
|
||||
"""Load all named features grouped by system and body."""
|
||||
results = []
|
||||
for table, ftype in [
|
||||
("atlas_cities", "city"),
|
||||
("atlas_rivers", "river"),
|
||||
("atlas_mountain_ranges", "mountain"),
|
||||
("atlas_oceans", "ocean"),
|
||||
("atlas_pois", "poi"),
|
||||
]:
|
||||
rows = conn.execute(f"""
|
||||
SELECT a.body_id, a.name, b.system_id,
|
||||
COALESCE(s.geographic_sector, 'unknown') as corridor,
|
||||
COALESCE(s.proper_name, s.system_id) as system_name,
|
||||
COALESCE(b.proper_name, '') as body_name,
|
||||
b.inhabited
|
||||
FROM {table} a
|
||||
JOIN bodies b ON a.body_id = b.body_id
|
||||
JOIN star_systems s ON b.system_id = s.system_id
|
||||
WHERE a.name IS NOT NULL AND a.name != ''
|
||||
""").fetchall()
|
||||
for body_id, name, system_id, corridor, sys_name, body_name, inhabited in rows:
|
||||
results.append({
|
||||
"body_id": body_id,
|
||||
"name": name,
|
||||
"system_id": system_id,
|
||||
"corridor": corridor,
|
||||
"system_name": sys_name,
|
||||
"body_name": body_name,
|
||||
"feature_type": ftype,
|
||||
"inhabited": bool(inhabited),
|
||||
})
|
||||
return results
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Checks
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def check_prompt_fragments(names):
|
||||
"""Find names that look like prompt leaks."""
|
||||
fragments = [
|
||||
"style:", "answer:", "insert your", "example", "placeholder",
|
||||
"number only", "names:", "generate", "already used", "do not",
|
||||
"must be", "distinct", "comma-separated", "best:", "option",
|
||||
]
|
||||
hits = []
|
||||
for n in names:
|
||||
low = n["name"].lower()
|
||||
for frag in fragments:
|
||||
if frag in low:
|
||||
hits.append((n["name"], n["body_id"], n["system_name"], frag))
|
||||
break
|
||||
return hits
|
||||
|
||||
|
||||
def check_exact_dupes_within_system(names):
|
||||
"""Find exact duplicate names within the same system + feature type."""
|
||||
by_sys_type = defaultdict(list)
|
||||
for n in names:
|
||||
key = (n["system_id"], n["feature_type"])
|
||||
by_sys_type[key].append(n)
|
||||
|
||||
dupes = []
|
||||
for key, group in by_sys_type.items():
|
||||
seen = {}
|
||||
for n in group:
|
||||
low = n["name"].lower()
|
||||
if low in seen:
|
||||
dupes.append((n["name"], n["body_id"], seen[low], n["system_name"], n["feature_type"]))
|
||||
else:
|
||||
seen[low] = n["body_id"]
|
||||
return dupes
|
||||
|
||||
|
||||
def check_exact_dupes_within_body(names):
|
||||
"""Find exact duplicate names within the same body across all types."""
|
||||
by_body = defaultdict(list)
|
||||
for n in names:
|
||||
by_body[n["body_id"]].append(n)
|
||||
|
||||
dupes = []
|
||||
for body_id, group in by_body.items():
|
||||
seen = {}
|
||||
for n in group:
|
||||
low = n["name"].lower()
|
||||
if low in seen:
|
||||
dupes.append((n["name"], body_id, n["feature_type"], seen[low], n["system_name"]))
|
||||
else:
|
||||
seen[low] = n["feature_type"]
|
||||
return dupes
|
||||
|
||||
|
||||
def check_body_name_echo(names):
|
||||
"""Find names where the body/system proper name dominates."""
|
||||
hits = []
|
||||
for n in names:
|
||||
if not n["body_name"]:
|
||||
continue
|
||||
body_stem = n["body_name"].lower()
|
||||
name_low = n["name"].lower()
|
||||
# Check if body name appears as a word in the feature name
|
||||
name_words = set(name_low.split())
|
||||
body_words = set(body_stem.split())
|
||||
if body_words & name_words:
|
||||
hits.append((n["name"], n["body_name"], n["body_id"], n["system_name"]))
|
||||
return hits
|
||||
|
||||
|
||||
def check_register_bleed(names):
|
||||
"""Find names that look like they're from the wrong cultural register.
|
||||
|
||||
Uses keyword heuristics — not perfect but catches obvious mismatches.
|
||||
"""
|
||||
# Register keywords that should NOT appear in certain corridors
|
||||
bleed_patterns = {
|
||||
"core": {
|
||||
"wrong": ["kimchi", "samurai", "fjord", "veld", "kopje", "baobab"],
|
||||
},
|
||||
"east_reach": {
|
||||
"wrong": ["bramble", "meadow", "thatch", "croft", "basilica", "forum", "senate"],
|
||||
},
|
||||
"west_reach": {
|
||||
"wrong": ["sakura", "bamboo", "lotus", "jade", "dragon", "phoenix"],
|
||||
},
|
||||
"south_reach": {
|
||||
"wrong": ["fjord", "viking", "norse", "highland", "glen"],
|
||||
},
|
||||
"north_reach": {
|
||||
"wrong": ["sakura", "bamboo", "jade", "polder", "graben"],
|
||||
},
|
||||
}
|
||||
# NZ/Australian names in non-core/north corridors
|
||||
nz_keywords = ["pōhutukawa", "waitara", "wairarapa", "fiordland", "aotearoa",
|
||||
"rangitoto", "wellington", "canterbury", "auckland", "otago",
|
||||
"kauri", "pukekohe", "taranaki", "moana"]
|
||||
|
||||
hits = []
|
||||
for n in names:
|
||||
low = n["name"].lower()
|
||||
corridor = n["corridor"]
|
||||
|
||||
# Check NZ bleed into non-Australian registers
|
||||
if corridor not in ("core", "north_reach"):
|
||||
for kw in nz_keywords:
|
||||
if kw in low:
|
||||
hits.append((n["name"], n["body_id"], corridor, n["system_name"],
|
||||
f"NZ/AU keyword '{kw}' in {corridor}"))
|
||||
break
|
||||
|
||||
# Check corridor-specific wrong keywords
|
||||
if corridor in bleed_patterns:
|
||||
for kw in bleed_patterns[corridor]["wrong"]:
|
||||
if kw in low:
|
||||
hits.append((n["name"], n["body_id"], corridor, n["system_name"],
|
||||
f"keyword '{kw}' wrong for {corridor}"))
|
||||
break
|
||||
return hits
|
||||
|
||||
|
||||
def check_feature_type_mismatch(names):
|
||||
"""Find names that sound wrong for their feature type."""
|
||||
# Street/road names shouldn't be mountains
|
||||
street_words = {"street", "avenue", "boulevard", "drive", "road", "lane",
|
||||
"way", "highway", "route", "thoroughfare"}
|
||||
# Building names shouldn't be rivers/oceans
|
||||
building_words = {"hall", "house", "building", "tower", "plaza", "square",
|
||||
"station", "terminal", "center", "centre"}
|
||||
|
||||
hits = []
|
||||
for n in names:
|
||||
words = set(n["name"].lower().split())
|
||||
if n["feature_type"] == "mountain" and words & street_words:
|
||||
hits.append((n["name"], n["body_id"], n["feature_type"], n["system_name"],
|
||||
f"street name as mountain"))
|
||||
if n["feature_type"] in ("river", "ocean") and words & building_words:
|
||||
# Allow "Hall" for classical register
|
||||
if n["corridor"] != "core":
|
||||
hits.append((n["name"], n["body_id"], n["feature_type"], n["system_name"],
|
||||
f"building name as {n['feature_type']}"))
|
||||
return hits
|
||||
|
||||
|
||||
def check_stem_repetition(names):
|
||||
"""Find bodies where too many features share the same first word."""
|
||||
by_body = defaultdict(list)
|
||||
for n in names:
|
||||
by_body[n["body_id"]].append(n)
|
||||
|
||||
hits = []
|
||||
for body_id, group in by_body.items():
|
||||
# Count first significant word per feature type
|
||||
by_type = defaultdict(list)
|
||||
for n in group:
|
||||
by_type[n["feature_type"]].append(n["name"])
|
||||
|
||||
for ftype, fnames in by_type.items():
|
||||
if len(fnames) < 4:
|
||||
continue
|
||||
first_words = [_words(name)[0] if _words(name) else "" for name in fnames]
|
||||
counts = Counter(first_words)
|
||||
for word, count in counts.most_common(3):
|
||||
if count >= 4 and word:
|
||||
hits.append((body_id, ftype, word, count, len(fnames),
|
||||
group[0]["system_name"]))
|
||||
return hits
|
||||
|
||||
|
||||
def check_short_long_names(names):
|
||||
"""Find very short (1 word, ≤3 chars) or very long names."""
|
||||
short = [(n["name"], n["body_id"], n["system_name"])
|
||||
for n in names if len(n["name"]) <= 3]
|
||||
long_ = [(n["name"], n["body_id"], n["system_name"])
|
||||
for n in names if len(n["name"]) > 40]
|
||||
return short, long_
|
||||
|
||||
|
||||
def check_numbers_in_names(names):
|
||||
"""Find names containing digits."""
|
||||
return [(n["name"], n["body_id"], n["system_name"])
|
||||
for n in names if re.search(r"\d", n["name"])]
|
||||
|
||||
|
||||
def corridor_distribution(names):
|
||||
"""Count names per corridor."""
|
||||
counts = Counter(n["corridor"] for n in names)
|
||||
return counts
|
||||
|
||||
|
||||
def feature_type_distribution(names):
|
||||
"""Count names per feature type."""
|
||||
counts = Counter(n["feature_type"] for n in names)
|
||||
return counts
|
||||
|
||||
|
||||
def coverage_gaps(conn):
|
||||
"""Find bodies with unnamed features."""
|
||||
gaps = []
|
||||
for table, ftype in [
|
||||
("atlas_mountain_ranges", "mountain"),
|
||||
("atlas_oceans", "ocean"),
|
||||
("atlas_pois", "poi"),
|
||||
]:
|
||||
rows = conn.execute(f"""
|
||||
SELECT a.body_id, COUNT(*) as total,
|
||||
SUM(CASE WHEN a.name IS NULL OR a.name = '' THEN 1 ELSE 0 END) as blank,
|
||||
COALESCE(s.proper_name, b.system_id) as sys_name,
|
||||
b.inhabited
|
||||
FROM {table} a
|
||||
JOIN bodies b ON a.body_id = b.body_id
|
||||
JOIN star_systems s ON b.system_id = s.system_id
|
||||
GROUP BY a.body_id
|
||||
HAVING blank > 0
|
||||
ORDER BY b.inhabited DESC, blank DESC
|
||||
""").fetchall()
|
||||
for body_id, total, blank, sys_name, inhabited in rows:
|
||||
gaps.append((body_id, ftype, blank, total, sys_name, bool(inhabited)))
|
||||
return gaps
|
||||
|
||||
|
||||
def most_common_names(names, top_n=20):
|
||||
"""Find the most frequently used names across all systems."""
|
||||
counts = Counter(n["name"].lower() for n in names)
|
||||
return counts.most_common(top_n)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Report
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
verbose = "--verbose" in sys.argv
|
||||
|
||||
conn = sqlite3.connect(str(DB_PATH), timeout=30.0)
|
||||
print("Loading named features from DB...")
|
||||
names = load_all_names(conn)
|
||||
print(f" {len(names):,} named features loaded\n")
|
||||
|
||||
# === Distribution ===
|
||||
print("=" * 70)
|
||||
print("DISTRIBUTION")
|
||||
print("=" * 70)
|
||||
|
||||
print("\nBy corridor:")
|
||||
for corridor, count in sorted(corridor_distribution(names).items(), key=lambda x: -x[1]):
|
||||
print(f" {corridor:20s} {count:>6,}")
|
||||
|
||||
print("\nBy feature type:")
|
||||
for ftype, count in sorted(feature_type_distribution(names).items(), key=lambda x: -x[1]):
|
||||
print(f" {ftype:20s} {count:>6,}")
|
||||
|
||||
# === Most common names ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("MOST COMMON NAMES (potential over-generation)")
|
||||
print("=" * 70)
|
||||
for name, count in most_common_names(names, 30):
|
||||
if count >= 3:
|
||||
print(f" {count:>4}x {name}")
|
||||
|
||||
# === Prompt fragments ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("PROMPT FRAGMENT LEAKS")
|
||||
print("=" * 70)
|
||||
fragments = check_prompt_fragments(names)
|
||||
if fragments:
|
||||
for name, body, sys_name, frag in fragments[:20]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\" (matched: {frag})")
|
||||
if len(fragments) > 20:
|
||||
print(f" ... and {len(fragments) - 20} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(fragments)}")
|
||||
|
||||
# === Exact dupes within system ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("EXACT DUPLICATES WITHIN SYSTEM (same name, same feature type)")
|
||||
print("=" * 70)
|
||||
sys_dupes = check_exact_dupes_within_system(names)
|
||||
if sys_dupes:
|
||||
for name, body1, body2, sys_name, ftype in sys_dupes[:20]:
|
||||
print(f" [{sys_name}] \"{name}\" ({ftype}) on {body1} and {body2}")
|
||||
if len(sys_dupes) > 20:
|
||||
print(f" ... and {len(sys_dupes) - 20} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(sys_dupes)}")
|
||||
|
||||
# === Exact dupes within body ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("EXACT DUPLICATES WITHIN BODY (same name, different feature types)")
|
||||
print("=" * 70)
|
||||
body_dupes = check_exact_dupes_within_body(names)
|
||||
if body_dupes:
|
||||
for name, body, ftype1, ftype2, sys_name in body_dupes[:20]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\" as {ftype1} and {ftype2}")
|
||||
if len(body_dupes) > 20:
|
||||
print(f" ... and {len(body_dupes) - 20} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(body_dupes)}")
|
||||
|
||||
# === Stem repetition ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("STEM REPETITION (4+ features sharing first word on same body)")
|
||||
print("=" * 70)
|
||||
stems = check_stem_repetition(names)
|
||||
if stems:
|
||||
for body, ftype, word, count, total, sys_name in stems[:20]:
|
||||
print(f" [{sys_name}/{body}] \"{word}\" appears {count}/{total} times in {ftype}s")
|
||||
if len(stems) > 20:
|
||||
print(f" ... and {len(stems) - 20} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(stems)}")
|
||||
|
||||
# === Body name echo ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("BODY NAME ECHO (planet name appears in feature name)")
|
||||
print("=" * 70)
|
||||
echoes = check_body_name_echo(names)
|
||||
if echoes:
|
||||
# Group by body
|
||||
by_body = defaultdict(list)
|
||||
for name, body_name, body_id, sys_name in echoes:
|
||||
by_body[(body_id, body_name, sys_name)].append(name)
|
||||
for (body_id, body_name, sys_name), echo_names in sorted(
|
||||
by_body.items(), key=lambda x: -len(x[1])
|
||||
)[:15]:
|
||||
print(f" [{sys_name}/{body_id}] body=\"{body_name}\": {', '.join(echo_names[:5])}"
|
||||
f"{'...' if len(echo_names) > 5 else ''} ({len(echo_names)} total)")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(echoes)} names across {len(set(e[2] for e in echoes))} bodies")
|
||||
|
||||
# === Register bleed ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("REGISTER BLEED (cultural mismatch for corridor)")
|
||||
print("=" * 70)
|
||||
bleeds = check_register_bleed(names)
|
||||
if bleeds:
|
||||
for name, body, corridor, sys_name, reason in bleeds[:30]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\" — {reason}")
|
||||
if len(bleeds) > 30:
|
||||
print(f" ... and {len(bleeds) - 30} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(bleeds)}")
|
||||
|
||||
# === Feature type mismatch ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("FEATURE TYPE MISMATCH (street names as mountains, etc.)")
|
||||
print("=" * 70)
|
||||
mismatches = check_feature_type_mismatch(names)
|
||||
if mismatches:
|
||||
for name, body, ftype, sys_name, reason in mismatches[:20]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\" — {reason}")
|
||||
if len(mismatches) > 20:
|
||||
print(f" ... and {len(mismatches) - 20} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(mismatches)}")
|
||||
|
||||
# === Short/long names ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("SHORT NAMES (≤3 chars)")
|
||||
print("=" * 70)
|
||||
short, long_ = check_short_long_names(names)
|
||||
if short:
|
||||
for name, body, sys_name in short[:15]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\"")
|
||||
if len(short) > 15:
|
||||
print(f" ... and {len(short) - 15} more")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(short)}")
|
||||
|
||||
print(f"\n{'=' * 70}")
|
||||
print("LONG NAMES (>40 chars)")
|
||||
print("=" * 70)
|
||||
if long_:
|
||||
for name, body, sys_name in long_[:15]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\"")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(long_)}")
|
||||
|
||||
# === Numbers in names ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("NUMBERS IN NAMES")
|
||||
print("=" * 70)
|
||||
numbered = check_numbers_in_names(names)
|
||||
if numbered:
|
||||
for name, body, sys_name in numbered[:15]:
|
||||
print(f" [{sys_name}/{body}] \"{name}\"")
|
||||
else:
|
||||
print(" None found ✓")
|
||||
print(f" Total: {len(numbered)}")
|
||||
|
||||
# === Coverage gaps ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("COVERAGE GAPS (bodies with unnamed features)")
|
||||
print("=" * 70)
|
||||
gaps = coverage_gaps(conn)
|
||||
inhabited_gaps = [g for g in gaps if g[5]]
|
||||
uninhabited_gaps = [g for g in gaps if not g[5]]
|
||||
if inhabited_gaps:
|
||||
print(f"\n INHABITED bodies with gaps ({len(inhabited_gaps)}):")
|
||||
for body, ftype, blank, total, sys_name, _ in inhabited_gaps[:10]:
|
||||
print(f" [{sys_name}/{body}] {blank}/{total} {ftype}s unnamed")
|
||||
if uninhabited_gaps:
|
||||
print(f"\n Uninhabited bodies with gaps ({len(uninhabited_gaps)}):")
|
||||
gap_by_type = Counter(g[1] for g in uninhabited_gaps)
|
||||
for ftype, count in gap_by_type.most_common():
|
||||
total_blank = sum(g[2] for g in uninhabited_gaps if g[1] == ftype)
|
||||
print(f" {ftype}: {count} bodies, {total_blank} unnamed features")
|
||||
|
||||
# === Summary ===
|
||||
print(f"\n{'=' * 70}")
|
||||
print("SUMMARY")
|
||||
print("=" * 70)
|
||||
total = len(names)
|
||||
issues = (len(fragments) + len(sys_dupes) + len(body_dupes) +
|
||||
len(stems) + len(bleeds) + len(mismatches) +
|
||||
len(short) + len(long_) + len(numbered))
|
||||
print(f" Total named features: {total:>8,}")
|
||||
print(f" Total QA issues found: {issues:>8,}")
|
||||
print(f" Issue rate: {issues/total*100:>7.2f}%")
|
||||
print(f" Prompt fragment leaks: {len(fragments):>8,}")
|
||||
print(f" Exact dupes (system): {len(sys_dupes):>8,}")
|
||||
print(f" Exact dupes (body): {len(body_dupes):>8,}")
|
||||
print(f" Stem repetition: {len(stems):>8,}")
|
||||
print(f" Body name echo: {len(echoes):>8,}")
|
||||
print(f" Register bleed: {len(bleeds):>8,}")
|
||||
print(f" Feature type mismatch: {len(mismatches):>8,}")
|
||||
print(f" Short names: {len(short):>8,}")
|
||||
print(f" Long names: {len(long_):>8,}")
|
||||
print(f" Numbers in names: {len(numbered):>8,}")
|
||||
print(f" Coverage gaps (inhabited):{len(inhabited_gaps):>8,}")
|
||||
print(f" Coverage gaps (uninh.): {len(uninhabited_gaps):>8,}")
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -31,8 +31,8 @@ mkdir -p "$LOG_DIR"
|
||||
STAMP="$(date +%Y%m%d-%H%M%S)"
|
||||
LOG="$LOG_DIR/atlas-naming-$STAMP.log"
|
||||
|
||||
BIN="$HOME/Projects/settled-reach/binaries/sr-voice-rocm"
|
||||
MODEL="/var/mnt/data/projects/settled-reach/main/server/models/gemma2.gguf"
|
||||
BIN="$HOME/Projects/settled-reach/binaries/sr-voice-tooling"
|
||||
MODEL="$HOME/Projects/settled-reach/models/gemma-4.gguf"
|
||||
DISTROBOX_NAME="reach-build"
|
||||
|
||||
if [[ ! -x "$BIN" ]]; then
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test batch naming: exercises naming_core against real Gemma 4.
|
||||
|
||||
Usage:
|
||||
python3 tooling/planet-gen/test_batch_naming.py
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import hashlib
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
from gemma_naming import (
|
||||
CORRIDOR_SUBSTYLES,
|
||||
DEFAULT_SUBSTYLES,
|
||||
_PROMPT_CONFIG,
|
||||
)
|
||||
from naming_core import (
|
||||
build_batch_prompt,
|
||||
parse_batch_response,
|
||||
select_distinct,
|
||||
)
|
||||
|
||||
BIN = Path.home() / "Projects/settled-reach/binaries/sr-voice-tooling"
|
||||
MODEL = Path.home() / "Projects/settled-reach/models/gemma-4.gguf"
|
||||
DISTROBOX = "reach-build"
|
||||
CTX_SIZE = 1024
|
||||
|
||||
# Full system simulations — 5 bodies each, accumulating taken list
|
||||
SYSTEM_SIMS = [
|
||||
{
|
||||
"name": "Ran", "corridor": "core",
|
||||
"inflection": "English countryside, rural, agricultural settlers",
|
||||
"hook": "RAN does not import food.",
|
||||
"bodies": [
|
||||
("GJ144b", "ambition"), ("GJ144c", "community"),
|
||||
("GJ144d", "fear"), ("GJ144e", "hope"), ("GJ144e-1", "loss"),
|
||||
],
|
||||
},
|
||||
{
|
||||
"name": "Groombridge", "corridor": "core",
|
||||
"inflection": "British colonial settlement era",
|
||||
"hook": "GROOMBRIDGE is where the money lives.",
|
||||
"bodies": [
|
||||
("GJ380b", "wealth"), ("GJ380c", "pride"),
|
||||
("GJ380d", "industry"), ("GJ380e", "ambition"), ("GJ380f", "fleeting"),
|
||||
],
|
||||
},
|
||||
{
|
||||
"name": "Cairnside", "corridor": "deep_frontier",
|
||||
"inflection": "frontier descriptive, geographic features named by surveyors",
|
||||
"hook": "CAIRNSIDE is a materials science program running for forty years.",
|
||||
"bodies": [
|
||||
("GJ892b", "defiance"), ("GJ892c", "isolation"),
|
||||
("GJ892d", "fear"), ("GJ892e", "hope"), ("GJ892f", "loss"),
|
||||
],
|
||||
},
|
||||
{
|
||||
"name": "Ratnagiri", "corridor": "north_reach",
|
||||
"inflection": "South African English settler",
|
||||
"hook": "RATNAGIRI has a monopoly on its primary export that no one engineered.",
|
||||
"bodies": [
|
||||
("GJ575Ab", "pride"), ("GJ575Ac", "community"),
|
||||
("GJ575Ad", "industry"), ("GJ575Ae", "devotion"), ("GJ575Af", "ambition"),
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
# Build test list
|
||||
TESTS = []
|
||||
for sim in SYSTEM_SIMS:
|
||||
for body_id, mood in sim["bodies"]:
|
||||
TESTS.append({
|
||||
"label": f"{sim['name']} — {body_id} (mood: {mood})",
|
||||
"system": sim["name"], "body": body_id, "corridor": sim["corridor"],
|
||||
"inflection": sim["inflection"],
|
||||
"feature_type": "mountain_range", "count": 8,
|
||||
"taken": f"__accumulate_{sim['name']}__",
|
||||
"hook": sim["hook"], "mood": mood,
|
||||
})
|
||||
|
||||
|
||||
def main():
|
||||
cmd = ["distrobox", "enter", DISTROBOX, "--",
|
||||
str(BIN),
|
||||
"--model", str(MODEL),
|
||||
"--ctx-size", str(CTX_SIZE)]
|
||||
print("starting sr-voice-tooling...", flush=True)
|
||||
proc = subprocess.Popen(
|
||||
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL, text=True, bufsize=1,
|
||||
start_new_session=True,
|
||||
)
|
||||
|
||||
accum: dict[str, list[str]] = {}
|
||||
|
||||
for test in TESTS:
|
||||
taken = test["taken"]
|
||||
if isinstance(taken, str) and taken.startswith("__accumulate_"):
|
||||
key = taken
|
||||
taken = list(accum.get(key, []))
|
||||
|
||||
# Primary batch
|
||||
prompt = build_batch_prompt(
|
||||
feature_type=test["feature_type"],
|
||||
inflection=test["inflection"],
|
||||
count=test["count"],
|
||||
taken=taken,
|
||||
prompt_config=_PROMPT_CONFIG,
|
||||
system_name=test["system"],
|
||||
body_name=test["body"],
|
||||
system_hook=test["hook"],
|
||||
mood=test.get("mood"),
|
||||
ctx_size=CTX_SIZE,
|
||||
)
|
||||
|
||||
seed = int(hashlib.sha256(
|
||||
f"batch|{test['body']}|{test['feature_type']}".encode()
|
||||
).hexdigest()[:8], 16)
|
||||
|
||||
req = json.dumps({"prompt": prompt, "seed": seed})
|
||||
proc.stdin.write(req + "\n")
|
||||
proc.stdin.flush()
|
||||
|
||||
t0 = time.time()
|
||||
resp_line = proc.stdout.readline()
|
||||
elapsed = time.time() - t0
|
||||
|
||||
try:
|
||||
resp = json.loads(resp_line)
|
||||
raw = resp.get("text", resp.get("error", ""))
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
raw = f"ERR: {resp_line!r}"
|
||||
|
||||
candidates = parse_batch_response(raw)
|
||||
selected = select_distinct(candidates, test["count"], taken)
|
||||
|
||||
# Refill from adjacent register if short
|
||||
if len(selected) < test["count"]:
|
||||
shortfall = test["count"] - len(selected)
|
||||
refill_taken = taken + selected
|
||||
|
||||
corridor = test.get("corridor", "core")
|
||||
substyles = CORRIDOR_SUBSTYLES.get(corridor, DEFAULT_SUBSTYLES)
|
||||
primary_idx = next(
|
||||
(i for i, s in enumerate(substyles)
|
||||
if s["inflection"] == test["inflection"]),
|
||||
0,
|
||||
)
|
||||
refill_idx = (primary_idx + 1) % len(substyles)
|
||||
refill_inflection = substyles[refill_idx]["inflection"]
|
||||
|
||||
refill_prompt = build_batch_prompt(
|
||||
feature_type=test["feature_type"],
|
||||
inflection=refill_inflection,
|
||||
count=shortfall * 3,
|
||||
taken=refill_taken,
|
||||
prompt_config=_PROMPT_CONFIG,
|
||||
system_name=test["system"],
|
||||
body_name=test["body"],
|
||||
system_hook=test["hook"],
|
||||
mood=test.get("mood"),
|
||||
ctx_size=CTX_SIZE,
|
||||
)
|
||||
refill_seed = int(hashlib.sha256(
|
||||
f"refill|{test['body']}|{test['feature_type']}".encode()
|
||||
).hexdigest()[:8], 16)
|
||||
|
||||
proc.stdin.write(json.dumps({"prompt": refill_prompt, "seed": refill_seed}) + "\n")
|
||||
proc.stdin.flush()
|
||||
t1 = time.time()
|
||||
refill_line = proc.stdout.readline()
|
||||
refill_elapsed = time.time() - t1
|
||||
|
||||
try:
|
||||
refill_resp = json.loads(refill_line)
|
||||
refill_raw = refill_resp.get("text", "")
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
refill_raw = ""
|
||||
|
||||
refill_candidates = parse_batch_response(refill_raw)
|
||||
extra = select_distinct(refill_candidates, shortfall, refill_taken)
|
||||
print(f" REFILL ({refill_inflection}): {len(refill_candidates)} cand → {len(extra)} new: {extra}")
|
||||
selected.extend(extra)
|
||||
|
||||
print(f" {test['label']}")
|
||||
print(f" {len(candidates)} cand → {len(selected)} selected ({elapsed:.1f}s) taken={len(taken)}")
|
||||
print(f" {selected}")
|
||||
|
||||
if isinstance(test["taken"], str) and test["taken"].startswith("__accumulate_"):
|
||||
key = test["taken"]
|
||||
accum.setdefault(key, []).extend(selected)
|
||||
print(f" [{test['system']}: {len(accum[key])} total]")
|
||||
|
||||
proc.stdin.close()
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
|
||||
except Exception:
|
||||
pass
|
||||
proc.wait()
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,163 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Quick test: send register-selection prompts to sr-voice for a few
|
||||
systems and print what Gemma actually picks.
|
||||
|
||||
Usage:
|
||||
python3 tooling/planet-gen/test_register_selection.py
|
||||
"""
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import hashlib
|
||||
import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
from gemma_naming import (
|
||||
load_wiki_context,
|
||||
CORRIDOR_SUBSTYLES,
|
||||
DEFAULT_SUBSTYLES,
|
||||
palette_for,
|
||||
_extract_cultural_lines,
|
||||
)
|
||||
|
||||
BIN = Path.home() / "Projects/settled-reach/binaries/sr-voice-tooling"
|
||||
MODEL = Path.home() / "Projects/settled-reach/models/gemma-4.gguf"
|
||||
DISTROBOX = "reach-build"
|
||||
CTX_SIZE = 1024
|
||||
|
||||
TEST_SYSTEMS = [
|
||||
("GJ 411", "south_reach", "Lalande — Iberian/Portuguese"),
|
||||
("GJ 884", "south_reach", "Matamba — Angolan/Lusophone"),
|
||||
("GJ 506", "west_reach", "Dokkum — Dutch/Frisian"),
|
||||
("GJ 581", "west_reach", "Rødvik — Nordic"),
|
||||
("GJ 34B", "east_reach", "Yongjin — Korean"),
|
||||
("GJ 205", "east_reach", "Kurashiki — Japanese"),
|
||||
("GJ 71", "core", "Gateway — administrative hub"),
|
||||
("GJ 144", "core", "Ran — agricultural"),
|
||||
]
|
||||
|
||||
|
||||
def build_prompt(system_id: str, corridor: str) -> tuple[str, list[dict]]:
|
||||
"""Build the register selection prompt. Returns (prompt, substyles)."""
|
||||
substyles = CORRIDOR_SUBSTYLES.get(corridor, DEFAULT_SUBSTYLES)
|
||||
wiki_text, gttr_text = load_wiki_context(system_id)
|
||||
|
||||
options = []
|
||||
for idx, style in enumerate(substyles, 1):
|
||||
options.append(f"{idx}. {style['inflection']}")
|
||||
option_block = "\n".join(options)
|
||||
|
||||
context_parts = []
|
||||
if gttr_text:
|
||||
for para in gttr_text.strip().split("\n\n"):
|
||||
cleaned = para.replace("#", "").strip()
|
||||
if cleaned.startswith("THE DRIFTER") or cleaned.startswith("DRIFTER"):
|
||||
continue
|
||||
if not cleaned or len(cleaned) < 20:
|
||||
continue
|
||||
context_parts.append(cleaned[:300])
|
||||
break
|
||||
if wiki_text:
|
||||
cultural = _extract_cultural_lines(wiki_text)
|
||||
if cultural:
|
||||
context_parts.append(cultural)
|
||||
|
||||
context = "\n".join(context_parts)
|
||||
|
||||
preamble = (
|
||||
"Match the star system to the best cultural naming register.\n\n"
|
||||
"System: Neustadt — German-heritage industrial town, west corridor, orderly municipal governance.\n"
|
||||
"1. German settlement 2. Dutch colonial 3. Nordic 4. Polish/Czech 5. Baltic/Finnish\n"
|
||||
"Best: 1\n\n"
|
||||
"System: Matsue — Japanese precision manufacturing hub, east corridor.\n"
|
||||
"1. Korean 2. Japanese 3. Taiwanese/Hakka 4. Filipino 5. Mixed East Asian\n"
|
||||
"Best: 2\n\n"
|
||||
"System: "
|
||||
)
|
||||
tail = f"\n{option_block}\nBest (number only):"
|
||||
|
||||
max_prompt_chars = (CTX_SIZE - 16) * 4
|
||||
budget = max_prompt_chars - len(preamble) - len(tail)
|
||||
if budget < 100:
|
||||
budget = 100
|
||||
if len(context) > budget:
|
||||
context = context[:budget]
|
||||
|
||||
prompt = f"{preamble}{context}{tail}"
|
||||
return prompt, substyles
|
||||
|
||||
|
||||
def main():
|
||||
# Show prompt sizes first
|
||||
print("Prompt token estimates (rough: chars/4):")
|
||||
for system_id, corridor, note in TEST_SYSTEMS:
|
||||
prompt, _ = build_prompt(system_id, corridor)
|
||||
est_tokens = len(prompt) // 4
|
||||
print(f" {system_id:<8s} ~{est_tokens:>4d} tokens ({len(prompt)} chars) {note}")
|
||||
print()
|
||||
|
||||
cmd = ["distrobox", "enter", DISTROBOX, "--",
|
||||
str(BIN),
|
||||
"--model", str(MODEL),
|
||||
"--ctx-size", str(CTX_SIZE)]
|
||||
print(f"starting sr-voice (ctx_size={CTX_SIZE})...", flush=True)
|
||||
proc = subprocess.Popen(
|
||||
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL, text=True, bufsize=1,
|
||||
)
|
||||
|
||||
results = []
|
||||
for system_id, corridor, note in TEST_SYSTEMS:
|
||||
prompt, substyles = build_prompt(system_id, corridor)
|
||||
seed = int(hashlib.sha256(
|
||||
f"register|{system_id}|0".encode()
|
||||
).hexdigest()[:8], 16)
|
||||
|
||||
req = json.dumps({"prompt": prompt, "seed": seed})
|
||||
proc.stdin.write(req + "\n")
|
||||
proc.stdin.flush()
|
||||
|
||||
t0 = time.time()
|
||||
resp_line = proc.stdout.readline()
|
||||
elapsed = time.time() - t0
|
||||
|
||||
try:
|
||||
resp = json.loads(resp_line)
|
||||
raw = resp.get("text", resp.get("error", ""))
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
raw = f"ERR:{resp_line!r}"
|
||||
|
||||
# Parse
|
||||
digits = re.search(r"\d+", raw.strip() or "")
|
||||
if digits:
|
||||
try:
|
||||
choice = int(digits.group())
|
||||
if 1 <= choice <= len(substyles):
|
||||
picked = f"#{choice} {substyles[choice - 1]['inflection']}"
|
||||
else:
|
||||
picked = f"OUT OF RANGE ({choice})"
|
||||
except ValueError:
|
||||
picked = "PARSE FAIL"
|
||||
else:
|
||||
picked = f"FAIL: {raw[:60]}"
|
||||
|
||||
hash_pal = palette_for(corridor, system_id)
|
||||
results.append((system_id, note, raw.strip()[:12], picked,
|
||||
hash_pal["inflection"], elapsed))
|
||||
|
||||
proc.stdin.close()
|
||||
proc.wait()
|
||||
|
||||
print()
|
||||
print(f"{'System':<8s} {'Raw':<12s} {'Gemma picked':<48s} {'Hash fallback':<45s} {'Time':>5s}")
|
||||
print("-" * 130)
|
||||
for system_id, note, raw, picked, hash_pick, elapsed in results:
|
||||
print(f"{system_id:<8s} {raw:<12s} {picked:<48s} {hash_pick:<45s} {elapsed:4.1f}s")
|
||||
print(f" └ {note}")
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user