Timmy_Foundation/the-nexus
15
1
Fork 2
Code Issues 89 Pull Requests 21 Actions Packages Projects Releases Wiki Activity

Add plugin system + world tick

Browse Source 
Plugin base class + registry + directory loader
WorldTickSystem: 60s tick, weather, time of day, room state evolution
Forge fire decay, Garden growth, Bridge rain
This commit is contained in:
Alexander Whitestone
2026-04-12 21:04:39 -04:00
parent 960c2248be
commit f7e21464e5
1 changed files with 333 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

335
multi_user_bridge.py
View File

@@ -33,6 +33,131 @@ from pathlib import Path
from datetime import datetime
from typing import Optional
# ── Plugin System ──────────────────────────────────────────────────────
class Plugin:
"""Base class for bridge plugins. Override methods to add game mechanics."""
name: str = "unnamed"
description: str = ""
def on_message(self, user_id: str, message: str, room: str) -> str | None:
"""Called on every chat message. Return a string to override the response,
or None to let the normal AI handle it."""
return None
def on_join(self, user_id: str, room: str) -> str | None:
"""Called when a player joins a room. Return a message to broadcast, or None."""
return None
def on_leave(self, user_id: str, room: str) -> str | None:
"""Called when a player leaves a room. Return a message to broadcast, or None."""
return None
def on_command(self, user_id: str, command: str, args: str, room: str) -> dict | None:
"""Called on MUD commands. Return a result dict to override command handling,
or None to let the default parser handle it."""
return None
class PluginRegistry:
"""Registry that manages plugin lifecycle and dispatches hooks."""
def __init__(self):
self._plugins: dict[str, Plugin] = {}
self._lock = threading.Lock()
def register(self, plugin: Plugin):
"""Register a plugin by its name."""
with self._lock:
self._plugins[plugin.name] = plugin
print(f"[PluginRegistry] Registered plugin: {plugin.name}")
def unregister(self, name: str) -> bool:
"""Unregister a plugin by name."""
with self._lock:
if name in self._plugins:
del self._plugins[name]
print(f"[PluginRegistry] Unregistered plugin: {name}")
return True
return False
def get(self, name: str) -> Plugin | None:
"""Get a plugin by name."""
return self._plugins.get(name)
def list_plugins(self) -> list[dict]:
"""List all registered plugins."""
return [
{"name": p.name, "description": p.description}
for p in self._plugins.values()
]
def fire_on_message(self, user_id: str, message: str, room: str) -> str | None:
"""Fire on_message hooks. First non-None return wins."""
for plugin in self._plugins.values():
result = plugin.on_message(user_id, message, room)
if result is not None:
return result
return None
def fire_on_join(self, user_id: str, room: str) -> str | None:
"""Fire on_join hooks. Collect all non-None returns."""
messages = []
for plugin in self._plugins.values():
result = plugin.on_join(user_id, room)
if result is not None:
messages.append(result)
return "\n".join(messages) if messages else None
def fire_on_leave(self, user_id: str, room: str) -> str | None:
"""Fire on_leave hooks. Collect all non-None returns."""
messages = []
for plugin in self._plugins.values():
result = plugin.on_leave(user_id, room)
if result is not None:
messages.append(result)
return "\n".join(messages) if messages else None
def fire_on_command(self, user_id: str, command: str, args: str, room: str) -> dict | None:
"""Fire on_command hooks. First non-None return wins."""
for plugin in self._plugins.values():
result = plugin.on_command(user_id, command, args, room)
if result is not None:
return result
return None
def load_from_directory(self, plugin_dir: str):
"""Auto-load all Python files in a directory as plugins."""
plugin_path = Path(plugin_dir)
if not plugin_path.is_dir():
print(f"[PluginRegistry] Plugin directory not found: {plugin_dir}")
return
for py_file in plugin_path.glob("*.py"):
if py_file.name.startswith("_"):
continue
try:
module_name = f"plugins.{py_file.stem}"
spec = __import__("importlib").util.spec_from_file_location(module_name, py_file)
module = __import__("importlib").util.module_from_spec(spec)
spec.loader.exec_module(module)
# Look for Plugin subclasses in the module
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (isinstance(attr, type)
and issubclass(attr, Plugin)
and attr is not Plugin):
plugin_instance = attr()
self.register(plugin_instance)
except Exception as e:
print(f"[PluginRegistry] Failed to load {py_file.name}: {e}")
plugin_registry = PluginRegistry()
# ── Crisis Protocol ────────────────────────────────────────────────────
CRISIS_PROTOCOL = """
@@ -604,9 +729,15 @@ class BridgeHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/bridge/health':
state_file = WORLD_DIR / 'world_state.json'
ws = json.loads(state_file.read_text()) if state_file.exists() else {}
self._json_response({
"status": "ok",
"active_sessions": session_manager.get_session_count(),
"tick": ws.get("tick", 0),
"time_of_day": ws.get("time_of_day"),
"weather": ws.get("weather"),
"world_tick_running": world_tick_system._running,
"timestamp": datetime.now().isoformat(),
})
elif self.path == '/bridge/sessions':
@@ -758,7 +889,17 @@ class BridgeHandler(BaseHTTPRequestHandler):
room, "player_join",
f"{username} has entered {room}.",
exclude_user=user_id, data=enter_event)
response = session.chat(message)
# Plugin hook: on_join
plugin_msg = plugin_registry.fire_on_join(user_id, room)
if plugin_msg:
notification_manager.notify(user_id, "plugin", plugin_msg, room=room, username=username)
# Plugin hook: on_message (can override response)
plugin_override = plugin_registry.fire_on_message(user_id, message, room)
if plugin_override is not None:
response = plugin_override
else:
response = session.chat(message)
# Auto-notify: crisis detection — scan response for crisis protocol keywords
crisis_keywords = ["988", "741741", "safe right now", "crisis", "Crisis Text Line"]
@@ -797,6 +938,12 @@ class BridgeHandler(BaseHTTPRequestHandler):
old_room, "player_leave",
f"{session.username} has left {old_room}.",
exclude_user=user_id, data=leave_event)
# Plugin hook: on_leave
plugin_leave = plugin_registry.fire_on_leave(user_id, old_room)
if plugin_leave:
notification_manager.broadcast_room(
old_room, "plugin", plugin_leave,
exclude_user=user_id)
# Enter new room
session.room = new_room
enter_event = presence_manager.enter_room(user_id, session.username, new_room)
@@ -1049,6 +1196,187 @@ class BridgeHandler(BaseHTTPRequestHandler):
pass # Suppress HTTP logs
# ── World Tick System ──────────────────────────────────────────────────
import random
class WorldTickSystem:
"""Advances the world state every TICK_INTERVAL seconds.
On each tick:
- Increment tick counter
- Cycle time_of_day (dawn -> morning -> midday -> afternoon -> dusk -> night)
- Randomly change weather (clear, cloudy, foggy, light rain, heavy rain, storm)
- Evolve room states:
* The Forge: fire decays (blazing -> burning -> glowing -> embers -> cold)
* The Garden: growth advances (seeds -> sprouts -> growing -> blooming -> harvest)
* The Bridge: rain toggles
- Broadcast world events to connected players
- Save world state to disk
"""
TICK_INTERVAL = 60 # seconds
TIME_CYCLE = ["dawn", "morning", "midday", "afternoon", "dusk", "night"]
WEATHER_OPTIONS = [
"clear", "clear", "clear", # weighted toward clear
"cloudy", "cloudy",
"foggy",
"light rain",
"heavy rain",
"storm",
]
FIRE_STAGES = ["cold", "embers", "glowing", "burning", "blazing"]
GROWTH_STAGES = ["seeds", "sprouts", "growing", "blooming", "harvest"]
def __init__(self):
self._thread: threading.Thread | None = None
self._running = False
self._state_lock = threading.Lock()
# ── lifecycle ───────────────────────────────────────────────────────
def start(self):
"""Start the tick loop in a daemon thread."""
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._loop, daemon=True, name="world-tick")
self._thread.start()
print("[WorldTick] Started — ticking every {}s".format(self.TICK_INTERVAL))
def stop(self):
self._running = False
# ── internal loop ───────────────────────────────────────────────────
def _loop(self):
while self._running:
time.sleep(self.TICK_INTERVAL)
try:
self._tick()
except Exception as e:
print(f"[WorldTick] Error during tick: {e}")
def _tick(self):
"""Execute one world tick."""
state_file = WORLD_DIR / 'world_state.json'
if state_file.exists():
state = json.loads(state_file.read_text())
else:
state = {"tick": 0, "time_of_day": "morning", "weather": "clear", "rooms": {}}
events: list[str] = []
with self._state_lock:
# 1. Advance tick counter
state["tick"] = state.get("tick", 0) + 1
tick_num = state["tick"]
# 2. Time of day — advance every tick
current_time = state.get("time_of_day", "morning")
try:
idx = self.TIME_CYCLE.index(current_time)
except ValueError:
idx = 0
next_idx = (idx + 1) % len(self.TIME_CYCLE)
new_time = self.TIME_CYCLE[next_idx]
state["time_of_day"] = new_time
if new_time == "dawn":
events.append("The sun rises over The Tower. A new day begins.")
elif new_time == "dusk":
events.append("The sky darkens. Dusk settles over the world.")
elif new_time == "night":
events.append("Night falls. The green LED in The Tower pulses in the dark.")
# 3. Weather — change every 3 ticks
if tick_num % 3 == 0:
old_weather = state.get("weather", "clear")
new_weather = random.choice(self.WEATHER_OPTIONS)
state["weather"] = new_weather
if new_weather != old_weather:
events.append(f"The weather shifts to {new_weather}.")
# 4. Room states
rooms = state.get("rooms", {})
# The Forge — fire decays one stage every 2 ticks unless rekindled
forge = rooms.get("The Forge", {})
if forge:
fire = forge.get("fire_state", "cold")
untouched = forge.get("fire_untouched_ticks", 0) + 1
forge["fire_untouched_ticks"] = untouched
if untouched >= 2 and fire != "cold":
try:
fi = self.FIRE_STAGES.index(fire)
except ValueError:
fi = 0
if fi > 0:
forge["fire_state"] = self.FIRE_STAGES[fi - 1]
forge["fire_untouched_ticks"] = 0
if forge["fire_state"] == "embers":
events.append("The forge fire has died down to embers.")
elif forge["fire_state"] == "cold":
events.append("The forge fire has gone cold.")
rooms["The Forge"] = forge
# The Garden — growth advances one stage every 4 ticks
garden = rooms.get("The Garden", {})
if garden:
if tick_num % 4 == 0:
growth = garden.get("growth_stage", "seeds")
try:
gi = self.GROWTH_STAGES.index(growth)
except ValueError:
gi = 0
if gi < len(self.GROWTH_STAGES) - 1:
garden["growth_stage"] = self.GROWTH_STAGES[gi + 1]
events.append(f"The garden has advanced to {garden['growth_stage']}.")
else:
# Reset cycle
garden["growth_stage"] = "seeds"
events.append("The garden has been harvested. New seeds are planted.")
rooms["The Garden"] = garden
# The Bridge — rain state follows weather
bridge = rooms.get("The Bridge", {})
if bridge:
weather = state.get("weather", "clear")
is_raining = weather in ("light rain", "heavy rain", "storm")
was_raining = bridge.get("rain_active", False)
bridge["rain_active"] = is_raining
if is_raining and not was_raining:
events.append("Rain begins to fall on The Bridge.")
elif not is_raining and was_raining:
events.append("The rain on The Bridge lets up.")
rooms["The Bridge"] = bridge
state["rooms"] = rooms
state["last_updated"] = datetime.now().isoformat()
# 5. Save world state
state_file.write_text(json.dumps(state, indent=2))
# 6. Broadcast events to all occupied rooms (outside lock)
if events:
event_text = " ".join(events)
for room_name in list(presence_manager._rooms.keys()):
players = presence_manager.get_players_in_room(room_name)
if players:
notification_manager.broadcast_room(
room_name, "world_tick", event_text,
data={"tick": state["tick"], "time_of_day": state["time_of_day"],
"weather": state.get("weather"), "events": events},
)
print(f"[WorldTick] tick={state['tick']} time={state['time_of_day']} "
f"weather={state.get('weather')} events={len(events)}")
world_tick_system = WorldTickSystem()
# ── Main ───────────────────────────────────────────────────────────────
def main():
@@ -1069,7 +1397,10 @@ def main():
print(f" POST /bridge/move — Move user to room (triggers presence)")
print(f" POST /bridge/command — MUD command parser (look, go, examine, say, ask)")
print()
# Start world tick system
world_tick_system.start()
server = HTTPServer((BRIDGE_HOST, BRIDGE_PORT), BridgeHandler)
server.serve_forever()