Compare commits
6 Commits
490ef4291a
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a6d2063677 | ||
|
|
2623246b12 | ||
|
|
bff06a0edd | ||
|
|
8b8d9f0af9 | ||
|
|
894c17291d | ||
|
|
ef8bbca9a4 |
@@ -77,7 +77,7 @@ Pattern: `world-{environment}` — extends platform pattern with `world` functio
|
|||||||
|
|
||||||
## Deployment Topology
|
## Deployment Topology
|
||||||
|
|
||||||
Per v0.10 architecture + 2026-04-26 housekeeping clarification (treat dioscuri as empty; old nimmersky/SkyrimNet workloads being retired):
|
Per v0.10 architecture + 2026-04-26 housekeeping clarification (treat dioscuri as empty; old nimmersky workloads being retired):
|
||||||
|
|
||||||
| Component | Where | Why |
|
| Component | Where | Why |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
|
|||||||
File diff suppressed because one or more lines are too long
@@ -489,7 +489,7 @@ This is *Marx-in-the-schema applied to epistemics.* Information asymmetry is not
|
|||||||
- UI-toggle privacy → diegetic in-between dimension with lifeforce-cost
|
- UI-toggle privacy → diegetic in-between dimension with lifeforce-cost
|
||||||
- Single monolithic prompt context → three-tier knowledge stack with per-layer propagation policy
|
- Single monolithic prompt context → three-tier knowledge stack with per-layer propagation policy
|
||||||
- "Every NPC knows everything immediately" → paced canon-propagation with priority/scope/rate/ttl per row
|
- "Every NPC knows everything immediately" → paced canon-propagation with priority/scope/rate/ttl per row
|
||||||
- Cross-NPC memory bleed (Mantella/SkyrimNet failure-mode) → per-player local SQLite isolation atop v0.5 lemniscate-geometry foreclosure (two-layer defense)
|
- Cross-NPC memory bleed (Mantella-class failure-mode) → per-player local SQLite isolation atop v0.5 lemniscate-geometry foreclosure (two-layer defense)
|
||||||
|
|
||||||
## Runtime sampling knobs
|
## Runtime sampling knobs
|
||||||
|
|
||||||
|
|||||||
423
runtime-engine/LEMNISCATE-RUNTIME-Pseudo-code-001.py
Normal file
423
runtime-engine/LEMNISCATE-RUNTIME-Pseudo-code-001.py
Normal file
@@ -0,0 +1,423 @@
|
|||||||
|
# LEMNISCATE RUNTIME - Pseudo-code Implementation
|
||||||
|
# Companion to runtime-engine/architecture.md
|
||||||
|
|
||||||
|
from enum import Enum, auto
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Dict, List, Optional, Set, Callable
|
||||||
|
from abc import ABC, abstractmethod
|
||||||
|
import math
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# CORE PRIMITIVES - The shape of flow
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
class SlotState(Enum):
|
||||||
|
"""State of a slot in the lemniscate"""
|
||||||
|
EMPTY = auto()
|
||||||
|
OCCUPIED = auto()
|
||||||
|
EXITING = auto()
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class VerifierFlags:
|
||||||
|
"""Marker bits carried by slot-tokens, read at crossing"""
|
||||||
|
exit_eligible: bool = False
|
||||||
|
has_spoken_this_roundtrip: bool = False
|
||||||
|
mid_action: bool = False
|
||||||
|
goal_satisfied: bool = False
|
||||||
|
silence_eligible: bool = False
|
||||||
|
priority_pull: bool = False
|
||||||
|
|
||||||
|
def clear_transient(self):
|
||||||
|
"""Clear flags that reset each crossing"""
|
||||||
|
self.has_spoken_this_roundtrip = False
|
||||||
|
self.silence_eligible = False
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class SlotToken:
|
||||||
|
"""A token carried around the loop - represents an NPC in the zone"""
|
||||||
|
npc_id: str
|
||||||
|
state: SlotState = SlotState.OCCUPIED
|
||||||
|
flags: VerifierFlags = field(default_factory=VerifierFlags)
|
||||||
|
# Accumulator for gesture alignment (recursive lemniscate)
|
||||||
|
gesture_alignment_accumulator: Dict[str, float] = field(default_factory=dict)
|
||||||
|
|
||||||
|
def add_gesture_alignment(self, trait: str, value: float):
|
||||||
|
"""Add a gesture/trait value to the accumulator"""
|
||||||
|
self.gesture_alignment_accumulator[trait] = \
|
||||||
|
self.gesture_alignment_accumulator.get(trait, 0.0) + value
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# THE LEMNISCATE - Geometry as clock
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
class Lemniscate:
|
||||||
|
"""
|
||||||
|
The core runtime topology - a through-flow figure-eight.
|
||||||
|
Traversal IS the turn order.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, n_slots: int):
|
||||||
|
self.n_slots = n_slots
|
||||||
|
self.slots: List[Optional[SlotToken]] = [None] * n_slots
|
||||||
|
|
||||||
|
# The cursor - position on the lemniscate
|
||||||
|
# (loop_id, slot_index) - loop 0 = A, loop 1 = B
|
||||||
|
self.cursor_loop: int = 0
|
||||||
|
self.cursor_slot: int = 0
|
||||||
|
|
||||||
|
# Entry and exit queues
|
||||||
|
self.entry_line: List[SlotToken] = []
|
||||||
|
self.exit_line: List[SlotToken] = []
|
||||||
|
|
||||||
|
# The crossing is the ONLY synchronous event
|
||||||
|
self.crossing_callbacks: List[Callable] = []
|
||||||
|
|
||||||
|
def advance_cursor(self) -> bool:
|
||||||
|
"""
|
||||||
|
Advance cursor through the lemniscate.
|
||||||
|
Returns True if we reached the crossing.
|
||||||
|
"""
|
||||||
|
self.cursor_slot += 1
|
||||||
|
|
||||||
|
# Completed one loop
|
||||||
|
if self.cursor_slot >= self.n_slots:
|
||||||
|
self.cursor_slot = 0
|
||||||
|
self.cursor_loop = 1 - self.cursor_loop # Toggle A/B
|
||||||
|
return True # Reached crossing
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
def at_crossing(self) -> bool:
|
||||||
|
"""Cursor is between loops - at the synchronous crossing point"""
|
||||||
|
return self.cursor_slot == 0
|
||||||
|
|
||||||
|
def get_active_token(self) -> Optional[SlotToken]:
|
||||||
|
"""Get token at current cursor position"""
|
||||||
|
if self.cursor_slot < len(self.slots):
|
||||||
|
return self.slots[self.cursor_slot]
|
||||||
|
return None
|
||||||
|
|
||||||
|
def register_crossing_callback(self, callback: Callable):
|
||||||
|
"""Register a function to run at each crossing"""
|
||||||
|
self.crossing_callbacks.append(callback)
|
||||||
|
|
||||||
|
def trigger_crossing(self):
|
||||||
|
"""
|
||||||
|
The crossing event - all state updates resolve atomically.
|
||||||
|
This is the ONLY time the zone touches the global state bus.
|
||||||
|
"""
|
||||||
|
# 1. Run all registered callbacks (dialog turns, state updates)
|
||||||
|
for callback in self.crossing_callbacks:
|
||||||
|
callback()
|
||||||
|
|
||||||
|
# 2. Process exit-eligible tokens
|
||||||
|
self._process_exits()
|
||||||
|
|
||||||
|
# 3. Push from entry line to fill vacancies
|
||||||
|
self._process_entries()
|
||||||
|
|
||||||
|
# 4. Clear transient flags
|
||||||
|
self._clear_transient_flags()
|
||||||
|
|
||||||
|
def _process_exits(self):
|
||||||
|
"""Route exit-eligible tokens to exit line"""
|
||||||
|
for i, token in enumerate(self.slots):
|
||||||
|
if token and token.flags.exit_eligible:
|
||||||
|
self.exit_line.append(token)
|
||||||
|
self.slots[i] = None
|
||||||
|
token.state = SlotState.EXITING
|
||||||
|
|
||||||
|
def _process_entries(self):
|
||||||
|
"""Fill vacant slots from entry line"""
|
||||||
|
for i, slot in enumerate(self.slots):
|
||||||
|
if slot is None and self.entry_line:
|
||||||
|
self.slots[i] = self.entry_line.pop(0)
|
||||||
|
|
||||||
|
def _clear_transient_flags(self):
|
||||||
|
"""Clear flags that reset each crossing"""
|
||||||
|
for token in self.slots:
|
||||||
|
if token:
|
||||||
|
token.flags.clear_transient()
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# ZONES - Bounded places of structured speaking
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
class ZoneRegister(Enum):
|
||||||
|
"""Which ontological register this zone lives in"""
|
||||||
|
PHYSICAL = auto()
|
||||||
|
LIMINAL = auto()
|
||||||
|
IMPERIAL = auto()
|
||||||
|
|
||||||
|
class Zone:
|
||||||
|
"""
|
||||||
|
A bounded, named, slot-indexed, director-managed event instance.
|
||||||
|
The zone IS the lemniscate, not a container for it.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, name: str, n_slots: int, register: ZoneRegister = ZoneRegister.PHYSICAL):
|
||||||
|
self.name = name
|
||||||
|
self.register = register
|
||||||
|
self.lemniscate = Lemniscate(n_slots)
|
||||||
|
|
||||||
|
# Purpose - spawn intent, immutable
|
||||||
|
self.purpose: str = ""
|
||||||
|
|
||||||
|
# State tracking
|
||||||
|
self.scene_state_public: Dict = {}
|
||||||
|
self.ternary_gate_edges: Dict[tuple, str] = {} # (npc_id, npc_id) -> CLOSED/STABLE/OPEN
|
||||||
|
self.drift_state: Optional[str] = None # Delta from spawn intent
|
||||||
|
|
||||||
|
# Lifeforce binding
|
||||||
|
self.lifeforce_budget: float = 0.0
|
||||||
|
self.lifeforce_burned: float = 0.0
|
||||||
|
self.pulse_rate: float = 1.0 # Crossings per second
|
||||||
|
|
||||||
|
# Lifecycle
|
||||||
|
self.is_terminated: bool = False
|
||||||
|
self.crossing_count: int = 0
|
||||||
|
|
||||||
|
# Player gesture overlay - broadcasts to all active slots
|
||||||
|
self.pending_gestures: List[tuple] = [] # (trait, value)
|
||||||
|
|
||||||
|
def set_purpose(self, purpose: str):
|
||||||
|
"""Set the zone's spawn intent - immutable after this"""
|
||||||
|
self.purpose = purpose
|
||||||
|
|
||||||
|
def add_npc(self, npc_id: str, traits: Dict[str, float]):
|
||||||
|
"""Add an NPC to the entry line"""
|
||||||
|
token = SlotToken(npc_id=npc_id)
|
||||||
|
for trait, value in traits.items():
|
||||||
|
token.gesture_alignment_accumulator[trait] = value
|
||||||
|
self.lemniscate.entry_line.append(token)
|
||||||
|
|
||||||
|
def add_gesture(self, trait: str, value: float):
|
||||||
|
"""
|
||||||
|
Add a player gesture to the accumulator.
|
||||||
|
These integrate at the next crossing.
|
||||||
|
"""
|
||||||
|
self.pending_gestures.append((trait, value))
|
||||||
|
|
||||||
|
def broadcast_gestures_to_active(self):
|
||||||
|
"""
|
||||||
|
Phase-locked overlay broadcast - all active slots perceive
|
||||||
|
the same player gesture simultaneously.
|
||||||
|
"""
|
||||||
|
for token in self.lemniscate.slots:
|
||||||
|
if token:
|
||||||
|
for trait, value in self.pending_gestures:
|
||||||
|
token.add_gesture_alignment(trait, value)
|
||||||
|
self.pending_gestures.clear()
|
||||||
|
|
||||||
|
def run_crossing(self):
|
||||||
|
"""Execute one axis crossing - the core unit of zone time"""
|
||||||
|
self.crossing_count += 1
|
||||||
|
|
||||||
|
# Broadcast player gestures before the crossing
|
||||||
|
self.broadcast_gestures_to_active()
|
||||||
|
|
||||||
|
# Trigger the crossing - all state updates resolve atomically
|
||||||
|
self.lemniscate.trigger_crossing()
|
||||||
|
|
||||||
|
# Lifeforce burn
|
||||||
|
active_slots = sum(1 for t in self.lemniscate.slots if t is not None)
|
||||||
|
self.lifeforce_burned += self.pulse_rate * active_slots
|
||||||
|
|
||||||
|
# Check for termination
|
||||||
|
if self.lifeforce_burned >= self.lifeforce_budget:
|
||||||
|
self.terminal_report("budget_exhausted")
|
||||||
|
self.is_terminated = True
|
||||||
|
return
|
||||||
|
|
||||||
|
# Empty queue behavior: shrink at every crossing
|
||||||
|
if len(self.lemniscate.entry_line) == 0:
|
||||||
|
self._shrink()
|
||||||
|
|
||||||
|
# Check if zone has dissolved
|
||||||
|
if all(t is None for t in self.lemniscate.slots):
|
||||||
|
self.terminal_report("dissolved")
|
||||||
|
self.is_terminated = True
|
||||||
|
|
||||||
|
def _shrink(self):
|
||||||
|
"""Dissolve one active slot when entry line is empty"""
|
||||||
|
# Find last occupied slot
|
||||||
|
for i in range(len(self.lemniscate.slots) - 1, -1, -1):
|
||||||
|
if self.lemniscate.slots[i] is not None:
|
||||||
|
self.lemniscate.exit_line.append(self.lemniscate.slots[i])
|
||||||
|
self.lemniscate.slots[i] = None
|
||||||
|
break
|
||||||
|
|
||||||
|
def terminal_report(self, reason: str) -> Dict:
|
||||||
|
"""Generate the report sent back to spawning director"""
|
||||||
|
participants = [t.npc_id for t in self.lemniscate.slots if t is not None]
|
||||||
|
return {
|
||||||
|
"name": self.name,
|
||||||
|
"purpose": self.purpose,
|
||||||
|
"reason": reason,
|
||||||
|
"crossings": self.crossing_count,
|
||||||
|
"lifeforce_cost": self.lifeforce_burned,
|
||||||
|
"participants": participants,
|
||||||
|
"drift": self.drift_state
|
||||||
|
}
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# TWO-LAYER EXECUTION - State machine + lemniscate
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
class StateMachineLayer:
|
||||||
|
"""
|
||||||
|
Physical action layer - runs at simulation tick (fast).
|
||||||
|
Decoupled from lemniscate cursor.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.actions: Dict[str, Callable] = {}
|
||||||
|
self.state: Dict[str, Any] = {}
|
||||||
|
|
||||||
|
def register_action(self, name: str, callback: Callable):
|
||||||
|
"""Register a physical action"""
|
||||||
|
self.actions[name] = callback
|
||||||
|
|
||||||
|
def execute(self, npc_id: str, action_name: str):
|
||||||
|
"""Execute a physical action for an NPC"""
|
||||||
|
if action_name in self.actions:
|
||||||
|
self.actions[action_name](npc_id, self.state)
|
||||||
|
|
||||||
|
def get_status(self) -> Dict:
|
||||||
|
"""Snapshot of physical state for dialog layer"""
|
||||||
|
return self.state.copy()
|
||||||
|
|
||||||
|
class EmergentZone(Zone):
|
||||||
|
"""
|
||||||
|
A dramatic episode unit with goal evaluation.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, name: str, n_slots: int, purpose: str,
|
||||||
|
goal: Dict, budget: float):
|
||||||
|
super().__init__(name, n_slots)
|
||||||
|
self.set_purpose(purpose)
|
||||||
|
self.goal = goal
|
||||||
|
self.lifeforce_budget = budget
|
||||||
|
|
||||||
|
# Two-layer execution
|
||||||
|
self.state_machine = StateMachineLayer()
|
||||||
|
self.dialog_layer = self # Zone IS the dialog layer
|
||||||
|
|
||||||
|
# Goal tracking
|
||||||
|
self.goal_satisfied = None
|
||||||
|
self.goal_axis: str = ""
|
||||||
|
|
||||||
|
def execute_turn(self, dt: float):
|
||||||
|
"""
|
||||||
|
Execute both layers in synchronized-but-loose harmony.
|
||||||
|
Speech doesn't gate action; action doesn't block speech.
|
||||||
|
"""
|
||||||
|
# State machine layer (fast, at sim tick)
|
||||||
|
for action in self.state_machine.actions:
|
||||||
|
# Execute physical actions
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Dialog layer (at axis rate)
|
||||||
|
if self._should_crossing(dt):
|
||||||
|
self.run_crossing()
|
||||||
|
self._evaluate_goal()
|
||||||
|
|
||||||
|
def _should_crossing(self, dt: float) -> bool:
|
||||||
|
"""Determine if axis crossing should occur based on pulse rate"""
|
||||||
|
# Simplified - would track cumulative time in real impl
|
||||||
|
return dt >= (1.0 / self.pulse_rate)
|
||||||
|
|
||||||
|
def _evaluate_goal(self):
|
||||||
|
"""Ternary evaluation against trait axis"""
|
||||||
|
if self.goal.get("type") == "numeric":
|
||||||
|
# Simple threshold
|
||||||
|
threshold = self.goal["threshold"]
|
||||||
|
# Would check state machine variable
|
||||||
|
self.goal_satisfied = 1 if True else 0
|
||||||
|
elif self.goal.get("type") == "fuzzy":
|
||||||
|
# Trait-axis ternary evaluation
|
||||||
|
axis = self.goal["axis"]
|
||||||
|
# Would evaluate ternary gate state delta
|
||||||
|
self.goal_satisfied = 0 # +1 / 0 / -1
|
||||||
|
self.goal_axis = axis
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# DISTRIBUTED FUNDING - Director + participant lifeforce
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
class Director:
|
||||||
|
"""
|
||||||
|
Manages zones for a region. Owns spawn decisions and budget.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, region: str, budget: float):
|
||||||
|
self.region = region
|
||||||
|
self.lifeforce_budget = budget
|
||||||
|
self.active_zones: Dict[str, Zone] = {}
|
||||||
|
|
||||||
|
def spawn_zone(self, name: str, purpose: str, slots: int,
|
||||||
|
budget: float, params: Dict = {}) -> Zone:
|
||||||
|
"""
|
||||||
|
Spawn a new zone - pays spawn cost from own budget.
|
||||||
|
"""
|
||||||
|
zone = EmergentZone(name, slots, purpose, params.get("goal", {}), budget)
|
||||||
|
zone.lifeforce_budget = budget
|
||||||
|
self.lifeforce_budget -= budget # Director pays spawn cost
|
||||||
|
self.active_zones[name] = zone
|
||||||
|
return zone
|
||||||
|
|
||||||
|
def update_all_zones(self, dt: float):
|
||||||
|
"""Advance all active zones"""
|
||||||
|
for zone in list(self.active_zones.values()):
|
||||||
|
if not zone.is_terminated:
|
||||||
|
zone.run_crossing()
|
||||||
|
# Would collect terminal reports
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# EXAMPLE: A rescue emergent zone
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
def example_rescue_zone_setup():
|
||||||
|
"""
|
||||||
|
Demonstrates the rescue event pattern from the architecture.
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Director has a budget
|
||||||
|
director = Director("district_1", budget=1000.0)
|
||||||
|
|
||||||
|
# NPC A's limb is broken - emergent signal
|
||||||
|
# Director consumes signal and spawns rescue event
|
||||||
|
|
||||||
|
rescue_goal = {
|
||||||
|
"type": "numeric",
|
||||||
|
"threshold": 10.0, # limb_A >= 10%
|
||||||
|
"fallback": {
|
||||||
|
"type": "fuzzy",
|
||||||
|
"axis": "Sophrosyne",
|
||||||
|
"direction": "positive"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
rescue = director.spawn_zone(
|
||||||
|
name="rescue_in_marketplace",
|
||||||
|
purpose="medical_emergency",
|
||||||
|
slots=4,
|
||||||
|
budget=50.0,
|
||||||
|
params={"goal": rescue_goal}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Add NPCs who might help
|
||||||
|
rescue.add_npc("npc_patient", {"Philotes": 0.3, "Sophrosyne": 0.1})
|
||||||
|
rescue.add_npc("npc_bystander_1", {"Philotes": 0.7, "Sophrosyne": 0.5})
|
||||||
|
rescue.add_npc("npc_bystander_2", {"Philotes": 0.2, "Sophrosyne": 0.8})
|
||||||
|
|
||||||
|
# Player adds a gesture (helping)
|
||||||
|
rescue.add_gesture("Philotes", 0.8)
|
||||||
|
|
||||||
|
# Runtime tick
|
||||||
|
rescue.execute_turn(0.016) # 60fps
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
example_rescue_zone_setup()
|
||||||
@@ -172,7 +172,7 @@ driver_context(cursor_at_NPC_i, mode) = {
|
|||||||
|
|
||||||
**The `knowledge_stack` is layered**, not a single bucket. Universal world-canon, district-canon (regional), NPC_i's own primary memory, and — only if the character is in the in-between dimension — clasp memory. The retrieval layer enforces the dimensional cut; the LLM never has to reason about it. See §Local memory architecture in inference-and-memory/architecture.md for the layering, propagation policy, and clasp-as-Ring-A* privacy primitive. The classic multi-agent hallucination source ("why does Kalypso suddenly remember what Anaximander confided to Phoibe?") is *structurally foreclosed* by the geometry. Write-back goes only to NPC_i's slice; the lemniscate guarantees NPC_i is not cursor-active again for at least one full roundtrip — write-back has all the time it needs without race conditions.
|
**The `knowledge_stack` is layered**, not a single bucket. Universal world-canon, district-canon (regional), NPC_i's own primary memory, and — only if the character is in the in-between dimension — clasp memory. The retrieval layer enforces the dimensional cut; the LLM never has to reason about it. See §Local memory architecture in inference-and-memory/architecture.md for the layering, propagation policy, and clasp-as-Ring-A* privacy primitive. The classic multi-agent hallucination source ("why does Kalypso suddenly remember what Anaximander confided to Phoibe?") is *structurally foreclosed* by the geometry. Write-back goes only to NPC_i's slice; the lemniscate guarantees NPC_i is not cursor-active again for at least one full roundtrip — write-back has all the time it needs without race conditions.
|
||||||
|
|
||||||
This converts multi-NPC dialog from an **emergent-chaos problem** into a **bounded-cast scene problem with a typed runtime**. The Mantella / SkyrimNet failure-modes are foreclosed at the architecture level rather than the prompt level:
|
This converts multi-NPC dialog from an **emergent-chaos problem** into a **bounded-cast scene problem with a typed runtime**. The classic multi-agent failure-modes are foreclosed at the architecture level rather than the prompt level:
|
||||||
|
|
||||||
- **Cast = active-slots, period.** Geographic proximity does not equal participation
|
- **Cast = active-slots, period.** Geographic proximity does not equal participation
|
||||||
- **Cursor sequences turns deterministically.** No concurrent LLM-storm
|
- **Cursor sequences turns deterministically.** No concurrent LLM-storm
|
||||||
|
|||||||
1
tools/imperial-typewriter/.gitignore
vendored
Normal file
1
tools/imperial-typewriter/.gitignore
vendored
Normal file
@@ -0,0 +1 @@
|
|||||||
|
output/
|
||||||
6
tools/imperial-typewriter/test-words.csv
Normal file
6
tools/imperial-typewriter/test-words.csv
Normal file
@@ -0,0 +1,6 @@
|
|||||||
|
# test wordlist for imperial-typewriter — comments and blank lines are skipped
|
||||||
|
DAFIT
|
||||||
|
MOIRA
|
||||||
|
NIMMER
|
||||||
|
EROS
|
||||||
|
SOPHROSYNE
|
||||||
|
190
tools/imperial-typewriter/typewriter.py
Normal file
190
tools/imperial-typewriter/typewriter.py
Normal file
@@ -0,0 +1,190 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Imperial typewriter — compose imperial-tongue words from glyph PNGs.
|
||||||
|
|
||||||
|
Reads glyph assets from the closed studio repo; writes composed word PNGs
|
||||||
|
to an output directory for VLM-decoder training data generation.
|
||||||
|
|
||||||
|
Composition is glyph-cell-tight (no gap between glyphs) — matches the
|
||||||
|
in-game inscription-typing layout. The imperial machine always screams:
|
||||||
|
input is auto-uppercased before composition.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import csv
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
DEFAULT_STUDIO_ROOT = "/home/dafit/nimmerverse/studio.nimmerworld.eachpath.local"
|
||||||
|
DEFAULT_GENERATION = "03"
|
||||||
|
DEFAULT_RESOLUTION = "512x1024"
|
||||||
|
|
||||||
|
VOWELS = set("AEIOUY")
|
||||||
|
CONSONANTS = set("BCDFGHJKLMNPQRSTVWXZ")
|
||||||
|
DIGITS = set("0123456789")
|
||||||
|
|
||||||
|
CATEGORY_INFO = {
|
||||||
|
"vowel": ("vowels", "vowel"),
|
||||||
|
"consonant": ("consonant", "consonant"),
|
||||||
|
"number": ("numbers", "number"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def category_for(char: str) -> tuple[str, str]:
|
||||||
|
"""Return (category-dir-name, file-prefix) for a single character."""
|
||||||
|
if char in VOWELS:
|
||||||
|
return CATEGORY_INFO["vowel"]
|
||||||
|
if char in CONSONANTS:
|
||||||
|
return CATEGORY_INFO["consonant"]
|
||||||
|
if char in DIGITS:
|
||||||
|
return CATEGORY_INFO["number"]
|
||||||
|
raise ValueError(f"No imperial-tongue glyph for character: {char!r}")
|
||||||
|
|
||||||
|
|
||||||
|
def glyph_path(
|
||||||
|
char: str,
|
||||||
|
polarity: str,
|
||||||
|
studio_root: str,
|
||||||
|
generation: str = DEFAULT_GENERATION,
|
||||||
|
resolution: str = DEFAULT_RESOLUTION,
|
||||||
|
) -> Path:
|
||||||
|
cat_dir, prefix = category_for(char)
|
||||||
|
suffix = "_neg" if polarity == "negative" else ""
|
||||||
|
filename = f"{prefix}_{char}_{resolution}{suffix}.png"
|
||||||
|
return (
|
||||||
|
Path(studio_root)
|
||||||
|
/ "imperial-cult"
|
||||||
|
/ "assets"
|
||||||
|
/ "imperial-tongue"
|
||||||
|
/ cat_dir
|
||||||
|
/ generation
|
||||||
|
/ f"{prefix}_{char}"
|
||||||
|
/ filename
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def compose_word(
|
||||||
|
word: str,
|
||||||
|
polarity: str,
|
||||||
|
studio_root: str,
|
||||||
|
generation: str = DEFAULT_GENERATION,
|
||||||
|
resolution: str = DEFAULT_RESOLUTION,
|
||||||
|
) -> Image.Image:
|
||||||
|
glyphs = []
|
||||||
|
for char in word:
|
||||||
|
path = glyph_path(char, polarity, studio_root, generation, resolution)
|
||||||
|
if not path.exists():
|
||||||
|
raise FileNotFoundError(f"Missing glyph asset: {path}")
|
||||||
|
glyphs.append(Image.open(path).convert("RGBA"))
|
||||||
|
|
||||||
|
width = sum(g.width for g in glyphs)
|
||||||
|
height = max(g.height for g in glyphs)
|
||||||
|
composed = Image.new("RGBA", (width, height), (0, 0, 0, 0))
|
||||||
|
|
||||||
|
x = 0
|
||||||
|
for g in glyphs:
|
||||||
|
composed.paste(g, (x, 0), g)
|
||||||
|
x += g.width
|
||||||
|
|
||||||
|
return composed
|
||||||
|
|
||||||
|
|
||||||
|
def words_from_csv(path: Path) -> list[str]:
|
||||||
|
"""Read words from a CSV file. Takes the first column of each non-empty row.
|
||||||
|
Skips blank rows and rows whose first cell is empty or starts with '#'."""
|
||||||
|
words = []
|
||||||
|
with open(path, newline="") as f:
|
||||||
|
reader = csv.reader(f)
|
||||||
|
for row in reader:
|
||||||
|
if not row:
|
||||||
|
continue
|
||||||
|
cell = row[0].strip()
|
||||||
|
if not cell or cell.startswith("#"):
|
||||||
|
continue
|
||||||
|
words.append(cell)
|
||||||
|
return words
|
||||||
|
|
||||||
|
|
||||||
|
def render_one(
|
||||||
|
word: str,
|
||||||
|
output_dir: Path,
|
||||||
|
polarities: list[str],
|
||||||
|
studio_root: str,
|
||||||
|
generation: str,
|
||||||
|
resolution: str,
|
||||||
|
) -> None:
|
||||||
|
word = word.upper()
|
||||||
|
for pol in polarities:
|
||||||
|
img = compose_word(word, pol, studio_root, generation, resolution)
|
||||||
|
suffix = "_neg" if pol == "negative" else ""
|
||||||
|
out_path = output_dir / f"{word}{suffix}.png"
|
||||||
|
img.save(out_path)
|
||||||
|
print(f"wrote {out_path} ({img.width}x{img.height})")
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
parser = argparse.ArgumentParser(
|
||||||
|
description="Imperial typewriter — compose imperial-tongue words from glyph PNGs.",
|
||||||
|
)
|
||||||
|
source = parser.add_mutually_exclusive_group(required=True)
|
||||||
|
source.add_argument(
|
||||||
|
"word",
|
||||||
|
nargs="?",
|
||||||
|
help="A single Latin word to compose. Auto-uppercased (imperial machine always screams).",
|
||||||
|
)
|
||||||
|
source.add_argument(
|
||||||
|
"--csv",
|
||||||
|
type=Path,
|
||||||
|
help="Path to a CSV/wordlist file. First column of each non-empty, non-'#'-prefixed row is used as a word.",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--output",
|
||||||
|
type=Path,
|
||||||
|
default=Path("./output"),
|
||||||
|
help="Output directory (created if missing). Default: ./output",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--polarity",
|
||||||
|
choices=["positive", "negative", "both"],
|
||||||
|
default="both",
|
||||||
|
help="Which polarity to render. Default: both",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--studio-root",
|
||||||
|
default=DEFAULT_STUDIO_ROOT,
|
||||||
|
help=f"Path to studio repo root. Default: {DEFAULT_STUDIO_ROOT}",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--generation",
|
||||||
|
default=DEFAULT_GENERATION,
|
||||||
|
help=f"Resolution generation subdir. Default: {DEFAULT_GENERATION}",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--resolution",
|
||||||
|
default=DEFAULT_RESOLUTION,
|
||||||
|
help=f"Glyph resolution. Default: {DEFAULT_RESOLUTION}",
|
||||||
|
)
|
||||||
|
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
args.output.mkdir(parents=True, exist_ok=True)
|
||||||
|
polarities = ["positive", "negative"] if args.polarity == "both" else [args.polarity]
|
||||||
|
|
||||||
|
if args.csv:
|
||||||
|
words = words_from_csv(args.csv)
|
||||||
|
if not words:
|
||||||
|
print(f"no words found in {args.csv}", file=sys.stderr)
|
||||||
|
sys.exit(1)
|
||||||
|
print(f"composing {len(words)} word(s) from {args.csv}")
|
||||||
|
for word in words:
|
||||||
|
try:
|
||||||
|
render_one(word, args.output, polarities, args.studio_root, args.generation, args.resolution)
|
||||||
|
except (ValueError, FileNotFoundError) as e:
|
||||||
|
print(f" skip {word!r}: {e}", file=sys.stderr)
|
||||||
|
else:
|
||||||
|
render_one(args.word, args.output, polarities, args.studio_root, args.generation, args.resolution)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Reference in New Issue
Block a user