#!/usr/bin/env python3
"""Analog Town: Shortwave Social Simulator — Gradio Application.
A whimsical 'shortwave radio' interface for simulating how different
fictional personas might internally process a piece of news.
"""
import base64
import json
import os
import shutil
import traceback
from pathlib import Path
import gradio as gr
from schemas import Town, BroadcastEvent, AgentState, SimulationResult
from model_client import ModelClient, synthesize_speech
from simulator import Simulator
from export_hub import ExportManager
from theme import CUSTOM_CSS
from town_generator import generate_town
# ──────────────────────────────────────────────
# Configuration
# ──────────────────────────────────────────────
SAMPLE_TOWNS_DIR = Path(__file__).parent / "sample_towns"
OUTPUTS_DIR = Path(__file__).parent / "outputs"
SAFETY_DISCLAIMER = (
"⚠ Analog Town is a perspective-rehearsal tool. "
"It does not predict real people. "
"Use fictional, composite, or consent-based personas. "
"Do not use it to manipulate, profile, or target real individuals."
)
# ──────────────────────────────────────────────
# Helper functions
# ──────────────────────────────────────────────
def get_available_towns() -> list[str]:
"""List available sample town presets."""
if not SAMPLE_TOWNS_DIR.exists():
return []
return [f.stem for f in SAMPLE_TOWNS_DIR.glob("*.json")]
def load_town(name: str) -> Town:
"""Load a town from the sample_towns directory."""
path = SAMPLE_TOWNS_DIR / f"{name}.json"
with open(path) as f:
data = json.load(f)
return Town(**data)
_EMOTION_COLORS = {
"anger": "#ff4444",
"fear": "#ff8c00",
"hope": "#ffb347",
"trust": "#39ff14",
"curiosity": "#47b3ff",
"social_energy": "#b347ff",
}
_EMOTION_KEYS = list(_EMOTION_COLORS.keys())
def _dominant_emotion_color(state) -> str:
values = {k: getattr(state, k) for k in _EMOTION_KEYS}
dominant = max(values, key=values.get)
return _EMOTION_COLORS[dominant]
def format_town_map(town: Town, selected_agent_id: str | None = None, sim_result=None) -> str:
"""Format the 2D interactive town map with character sprites."""
if not town.agents:
return "No map data
"
transition_map = {}
if sim_result is not None:
for t in sim_result.transitions:
transition_map[t.agent_id] = t
sprites = []
for agent in town.agents:
pos = agent.map_pos or {"x": 50, "y": 50}
x = pos.get("x", 50)
y = pos.get("y", 50)
selected_class = "selected-sprite" if agent.id == selected_agent_id else ""
avatar_file = agent.avatar if agent.avatar else "avatars/default.png"
avatar_path = Path(__file__).parent / avatar_file
if not avatar_path.exists():
avatar_path = Path(__file__).parent / "avatars/default.png"
avatar_url = f"/gradio_api/file={avatar_path.resolve()}"
transition = transition_map.get(agent.id)
if transition is not None:
tint_color = _dominant_emotion_color(transition.updated_state)
avatar_style = (
f"background-image: url('{avatar_url}');"
f"border-color: {tint_color};"
f"box-shadow: 0 0 12px {tint_color};"
f"transition: border-color 0.4s ease, box-shadow 0.4s ease;"
)
has_anomaly = any(
abs(v) > 0.6 for v in transition.emotion_delta.values()
)
else:
avatar_style = f"background-image: url('{avatar_url}');"
has_anomaly = False
anomaly_html = '📢
' if has_anomaly else ""
sprites.append(f"""
{agent.name}
{anomaly_html}
""")
sprites_html = "\n".join(sprites)
map_filename = town.map_image if town.map_image else "town_map.png"
map_image_path = Path(__file__).parent / map_filename
if not map_image_path.exists():
map_image_path = Path(__file__).parent / "town_map.png"
return f"""
})
{sprites_html}
"""
def format_agent_profile_card(agent, town) -> str:
"""Format full agent characteristics, history, relationships as HTML."""
if not agent:
return "Select an agent on the map to inspect dossier
"
values_str = " ".join([f'{v}' for v in agent.core_values])
wounds_str = "".join([f'{w}' for w in agent.private_history])
fears_str = "".join([f'{f}' for f in agent.fears])
hopes_str = "".join([f'{h}' for h in agent.hopes])
# Relationships
rel_rows = []
agent_names = {a.id: a.name for a in town.agents}
for other_id, rel_desc in agent.relationships.items():
other_name = agent_names.get(other_id, other_id.title())
rel_rows.append(f"""
{other_name}: {rel_desc}
""")
rel_str = "\n".join(rel_rows) if rel_rows else "None documented
"
return f"""
{agent.name}
{agent.role} · Age: {agent.age_range}
📡 {agent.frequency:.1f} FM
"{agent.public_description}"
CORE VALUES
{values_str}
INTERPERSONAL DYNAMICS
{rel_str}
"""
def format_frequency_display(freq: float, agent_name: str | None, signal_strength: int) -> str:
"""Format the frequency display with signal bars."""
bars = ""
for i in range(5):
active = "active" if i < signal_strength else ""
bars += f''
name_text = agent_name if agent_name else "SCANNING..."
color = "#39ff14" if agent_name else "#5a5248"
return f"""
{freq:.1f} FM
{bars}
{name_text}
"""
def format_monologue(text: str | None, agent_name: str | None) -> str:
"""Format the intercepted monologue."""
if not text:
return """
krrrssh... zzzt... no clear signal...
···· adjust frequency ····
"""
return f"""
"{text}"
— intercepted from {agent_name}'s thoughtstream
"""
def format_emotion_bars(state: AgentState | None) -> str:
"""Format emotional state as visual bars."""
if not state:
return "No signal data
"
emotions = {
"Trust": (state.trust, "#39ff14"),
"Anger": (state.anger, "#ff4444"),
"Fear": (state.fear, "#ff8c00"),
"Hope": (state.hope, "#ffb347"),
"Curiosity": (state.curiosity, "#47b3ff"),
"Social": (state.social_energy, "#b347ff"),
}
rows = []
for name, (value, color) in emotions.items():
pct = int(value * 100)
rows.append(f"""
""")
return "\n".join(rows)
def format_action_display(transition) -> str:
"""Format the agent's likely actions."""
if not transition:
return "No action data
"
return f"""
PRIVATE: {transition.likely_private_action}
PUBLIC: {transition.likely_public_action}
CONFLICT: {transition.value_conflict}
UNCERTAINTY: {transition.uncertainty}
"""
def format_timeline_strip(timeline_json: str, active_day: int | None = None) -> str:
"""Return a horizontal row of clickable day pills from a JSON timeline."""
if not timeline_json:
return 'Broadcast to begin a day timeline ▸
Broadcast to begin a day timeline ▸
Broadcast to begin a day timeline ▸
'
f'DAY {day}'
f'{title}'
f'{anomaly_html}'
f'
'
)
pills_html = "\n".join(pills)
return f'{pills_html}
'
def format_day_badge(day: int | None) -> str:
"""Return an absolutely-positioned day badge HTML for the map panel."""
if day is None:
return ""
return f'DAY {day}
'
def _build_voice_player(monologue: str | None, agent_name: str | None = None, agent_age: str | None = None) -> str:
"""Return a hidden div whose data attributes carry the monologue to the browser TTS poller."""
import html as _html
if not monologue:
return ""
clean = monologue.replace("\n", " ").replace("\r", " ").strip()
safe_text = _html.escape(clean, quote=True)
safe_name = _html.escape((agent_name or ""), quote=True)
safe_age = _html.escape((agent_age or ""), quote=True)
return (
f''
)
def tune_frequency(freq_value, sim_result_json):
"""Tune to a frequency and return the nearest agent's data."""
_empty_voice = _build_voice_player(None)
if not sim_result_json:
return (
format_frequency_display(freq_value, None, 0),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
)
try:
result = SimulationResult(**json.loads(sim_result_json))
except Exception:
return (
format_frequency_display(freq_value, None, 0),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
)
if not result.transitions:
return (
format_frequency_display(freq_value, None, 0),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
)
# Load town to get agent info
try:
town = load_town(result.town_id)
agent_freq_map = {a.id: a for a in town.agents}
except Exception:
agent_freq_map = {}
# Find nearest agent by frequency
nearest_transition = None
nearest_freq = None
nearest_name = None
min_dist = float("inf")
for t in result.transitions:
agent = agent_freq_map.get(t.agent_id)
if agent:
dist = abs(agent.frequency - freq_value)
if dist < min_dist:
min_dist = dist
nearest_transition = t
nearest_freq = agent.frequency
nearest_name = agent.name
# Check if we're close enough to get a signal
threshold = 1.2
if min_dist > threshold or not nearest_transition:
signal_strength = max(0, int(5 - min_dist * 2)) if nearest_transition else 0
return (
format_frequency_display(freq_value, None, signal_strength),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
)
# Calculate signal strength (stronger when closer)
signal_strength = max(1, int(5 - min_dist * 4))
trace_json = json.dumps(nearest_transition.model_dump(), indent=2, default=str)
agent = agent_freq_map.get(nearest_transition.agent_id)
nearest_age = agent.age_range if agent else ""
voice_html = _build_voice_player(
nearest_transition.internal_monologue, nearest_name, nearest_age
)
return (
format_frequency_display(nearest_freq, nearest_name, signal_strength),
format_monologue(nearest_transition.internal_monologue, nearest_name),
format_emotion_bars(nearest_transition.updated_state),
format_action_display(nearest_transition),
trace_json,
voice_html,
)
# ──────────────────────────────────────────────
# Main event handlers
# ──────────────────────────────────────────────
def on_load_town(town_name):
"""Load a preset town and reset every town-scoped surface (timeline, receiver, badges)."""
empty_voice = _build_voice_player(None)
empty_freq = format_frequency_display(88.0, None, 0)
empty_monologue = format_monologue(None, None)
empty_emotions = format_emotion_bars(None)
empty_actions = format_action_display(None)
empty_strip = format_timeline_strip("")
empty_profile = "Select an agent on the map to inspect dossier
"
if not town_name:
return (
"", empty_profile, "", "", "",
"", "", empty_strip, "",
"",
empty_freq, empty_monologue, empty_emotions, empty_actions, "", empty_voice,
gr.update(value=88.0),
)
try:
town = load_town(town_name)
map_html = format_town_map(town)
town_json = town.model_dump_json(indent=2)
event_title = town.default_event.title if town.default_event else ""
event_content = town.default_event.content if town.default_event else ""
return (
map_html,
empty_profile,
town_json,
event_title,
event_content,
"",
"",
empty_strip,
"",
"",
empty_freq,
empty_monologue,
empty_emotions,
empty_actions,
"",
empty_voice,
gr.update(value=88.0),
)
except Exception as e:
return (
f"Error loading town: {e}
",
empty_profile, "", "", "",
"", "", empty_strip, "",
f"❌ {e}",
empty_freq, empty_monologue, empty_emotions, empty_actions, "", empty_voice,
gr.update(value=88.0),
)
def on_select_agent(agent_id, town_json, sim_result_json):
"""Event handler when an agent is selected via map click."""
_empty = _build_voice_player(None)
if not town_json:
return "", "", 88.0, "", "", "", "", "", _empty
try:
town = Town(**json.loads(town_json))
agent = next((a for a in town.agents if a.id == agent_id), None)
if not agent:
return "", "", 88.0, "", "", "", "", "", _empty
sim_result = None
if sim_result_json:
try:
sim_result = SimulationResult(**json.loads(sim_result_json))
except Exception:
sim_result = None
map_html = format_town_map(town, agent.id, sim_result)
profile_html = format_agent_profile_card(agent, town)
# Get simulated shortwave state (monologue, emotional bars, action, voice)
receiver_data = tune_frequency(agent.frequency, sim_result_json)
return (
map_html,
profile_html,
agent.frequency,
receiver_data[0],
receiver_data[1],
receiver_data[2],
receiver_data[3],
receiver_data[4],
receiver_data[5],
)
except Exception as e:
traceback.print_exc()
return "", f"Error: {e}
", 88.0, "", "", "", "", "", _empty
def on_transmit(town_json, event_title, event_content, timeline_json, progress=gr.Progress()):
"""Run the simulation, chaining from the last day's states if available."""
if not town_json:
return "⚠ Load a town first", "", gr.update(), timeline_json, format_timeline_strip(timeline_json), ""
if not event_content:
return "⚠ Enter a broadcast event", "", gr.update(), timeline_json, format_timeline_strip(timeline_json), ""
try:
town = Town(**json.loads(town_json))
except Exception as e:
return f"❌ Invalid town data: {e}", "", gr.update(), timeline_json, format_timeline_strip(timeline_json), ""
event = BroadcastEvent(
title=event_title or "Broadcast Event",
content=event_content,
source="Town Radio",
)
try:
client = ModelClient()
simulator = Simulator(model_client=client)
except Exception as e:
return f"❌ Model initialization failed: {e}", "", gr.update(), timeline_json, format_timeline_strip(timeline_json), ""
previous_states = None
current_timeline = []
if timeline_json:
try:
current_timeline = json.loads(timeline_json)
if current_timeline:
last_entry = current_timeline[-1]
last_result = SimulationResult(**last_entry["sim_result"])
previous_states = {
t.agent_id: t.updated_state for t in last_result.transitions
}
except Exception:
current_timeline = []
day = len(current_timeline) + 1
status_lines = []
def progress_callback(name, status, index, total):
emoji = {"processing": "⏳", "complete": "✅"}.get(status.split(":")[0], "⚠")
line = f"{emoji} [{index+1}/{total}] {name}: {status}"
status_lines.append(line)
progress((index + 1) / total, desc=f"Processing {name}...")
try:
result = simulator.run_town_simulation(
town=town,
event=event,
previous_states=previous_states,
progress_callback=progress_callback,
day=day,
)
except Exception as e:
return f"❌ Simulation failed: {e}", "", gr.update(), timeline_json, format_timeline_strip(timeline_json), ""
n_success = len(result.transitions)
n_total = len(town.agents)
if n_success == n_total:
status = f"✅ Signal acquired — {n_success}/{n_total} transmissions intercepted"
elif n_success > 0:
status = f"⚠ Partial signal — {n_success}/{n_total} transmissions intercepted"
else:
status = "❌ No signal — all transmissions failed"
status += "\n" + "\n".join(status_lines)
new_entry = {
"day": day,
"event_title": event_title or "Broadcast Event",
"event_content": event_content,
"sim_result": json.loads(result.model_dump_json()),
}
current_timeline.append(new_entry)
new_timeline_json = json.dumps(current_timeline)
result_json = result.model_dump_json(indent=2)
updated_map = format_town_map(town, None, result)
timeline_strip = format_timeline_strip(new_timeline_json, active_day=day)
day_badge = format_day_badge(day)
return status, result_json, updated_map, new_timeline_json, timeline_strip, day_badge
def on_reset_timeline(town_json):
"""Clear the entire timeline and return to Day 1 default state."""
empty_strip = format_timeline_strip("")
if not town_json:
return "", "", empty_strip, ""
try:
town = Town(**json.loads(town_json))
map_html = format_town_map(town, None, None)
except Exception:
map_html = ""
return "", map_html, empty_strip, ""
def on_select_timeline_day(day_str, town_json, timeline_json):
"""Restore the map and sim_result to a specific historical day."""
_empty_voice = _build_voice_player(None)
empty_returns = (
"",
"",
format_frequency_display(88.0, None, 0),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
format_timeline_strip(timeline_json),
"",
)
if not day_str or not timeline_json:
return empty_returns
try:
day = int(day_str)
timeline = json.loads(timeline_json)
except Exception:
return empty_returns
entry = next((e for e in timeline if e["day"] == day), None)
if not entry:
return empty_returns
try:
sim_result = SimulationResult(**entry["sim_result"])
town = Town(**json.loads(town_json)) if town_json else None
except Exception:
return empty_returns
result_json = sim_result.model_dump_json(indent=2)
map_html = format_town_map(town, None, sim_result) if town else ""
strip_html = format_timeline_strip(timeline_json, active_day=day)
badge_html = format_day_badge(day)
return (
result_json,
map_html,
format_frequency_display(88.0, None, 0),
format_monologue(None, None),
format_emotion_bars(None),
format_action_display(None),
"",
_empty_voice,
strip_html,
badge_html,
)
def on_export_local(sim_result_json, town_json):
"""Export simulation traces to a local ZIP file."""
if not sim_result_json:
return None, "⚠ Run a simulation first"
try:
result = SimulationResult(**json.loads(sim_result_json))
town = Town(**json.loads(town_json))
exporter = ExportManager()
zip_path = exporter.export_zip(result, town, str(OUTPUTS_DIR))
return zip_path, f"✅ Exported to {zip_path}"
except Exception as e:
return None, f"❌ Export failed: {e}"
def on_export_hub(sim_result_json, town_json, repo_id):
"""Upload traces to Hugging Face Hub."""
if not sim_result_json:
return "⚠ Run a simulation first"
if not repo_id:
return "⚠ Enter a repository ID (e.g., username/analog-town-traces)"
try:
result = SimulationResult(**json.loads(sim_result_json))
town = Town(**json.loads(town_json))
exporter = ExportManager()
url = exporter.export_to_hub(result, town, repo_id)
return f"✅ Uploaded to {url}"
except Exception as e:
return f"❌ Hub upload failed: {e}"
# ──────────────────────────────────────────────
# Creative Mode handlers
# ──────────────────────────────────────────────
def on_generate_creative_town(concept, n_agents, current_json):
"""Generate a town JSON from concept + count. Returns (json_str, status_markdown)."""
if not concept or not concept.strip():
return current_json, "_Status: ⚠ Type a concept first_"
try:
town_dict = generate_town(concept.strip(), int(n_agents))
return json.dumps(town_dict, indent=2), "_Status: ✅ Town generated — review and edit the JSON below, then click Save & Load_"
except Exception as e:
return current_json, f"_Status: ❌ Generation failed: {e}_"
def on_save_load_creative_town(json_text, map_upload_path):
"""Validate JSON, optionally copy uploaded map image, write to sample_towns, then trigger full reset."""
empty_voice = _build_voice_player(None)
empty_freq = format_frequency_display(88.0, None, 0)
empty_monologue = format_monologue(None, None)
empty_emotions = format_emotion_bars(None)
empty_actions = format_action_display(None)
empty_strip = format_timeline_strip("")
empty_profile = "Select an agent on the map to inspect dossier
"
def _error_return(msg):
return (
"", empty_profile, "", "", "",
"", "", empty_strip, "",
f"❌ {msg}",
empty_freq, empty_monologue, empty_emotions, empty_actions, "", empty_voice,
gr.update(value=88.0),
gr.update(),
f"_Status: ❌ {msg}_",
)
if not json_text or not json_text.strip():
return _error_return("No JSON to save. Generate a town first.")
try:
town_dict = json.loads(json_text)
town = Town(**town_dict)
except Exception as e:
return _error_return(f"Invalid town JSON: {e}")
project_root = Path(__file__).parent
if map_upload_path:
dest = project_root / f"{town.id}_map.png"
shutil.copy(map_upload_path, dest)
town.map_image = f"{town.id}_map.png"
else:
if town.map_image:
candidate = project_root / town.map_image
if not candidate.exists():
town.map_image = "town_map.png"
else:
town.map_image = "town_map.png"
save_path = SAMPLE_TOWNS_DIR / f"{town.id}.json"
SAMPLE_TOWNS_DIR.mkdir(parents=True, exist_ok=True)
save_path.write_text(town.model_dump_json(indent=2))
new_choices = get_available_towns()
load_tuple = on_load_town(town.id)
return load_tuple + (
gr.update(choices=new_choices, value=town.id),
f"_Status: ✅ Town '{town.name}' saved and loaded. Press TRANSMIT to run Day 1._",
)
# ──────────────────────────────────────────────
# Build UI
# ──────────────────────────────────────────────
CUSTOM_JS_TEMPLATE = """
window.selectAgent = function(agentId) {
console.log("selectAgent called with agentId:", agentId);
const container = document.getElementById("selected-agent-id");
if (container) {
const input = container.querySelector("input") || container.querySelector("textarea");
if (input) {
let prototype = Object.getPrototypeOf(input);
let descriptor = Object.getOwnPropertyDescriptor(prototype, "value");
while (prototype && !descriptor) {
prototype = Object.getPrototypeOf(prototype);
descriptor = Object.getOwnPropertyDescriptor(prototype, "value");
}
const setter = descriptor ? descriptor.set : null;
if (setter) {
setter.call(input, agentId);
} else {
input.value = agentId;
}
input.dispatchEvent(new Event("input", { bubbles: true }));
setTimeout(() => {
const btn = document.getElementById("select-agent-trigger");
if (btn) {
console.log("Clicking select-agent-trigger button");
btn.click();
} else {
console.error("select-agent-trigger button not found");
}
}, 50);
} else {
console.error("Input/textarea not found inside #selected-agent-id");
}
} else {
console.error("#selected-agent-id container not found");
}
};
let ambientAudio = null;
let ambientMuted = false;
let ambientVolume = 0.15;
let currentActualVolume = 0.15;
let targetActualVolume = 0.15;
window.initAmbient = function() {
if (ambientAudio) return;
console.log("Initializing ambient audio with url: __AUDIO_URL__");
ambientAudio = new Audio("__AUDIO_URL__");
ambientAudio.loop = true;
ambientAudio.volume = ambientVolume;
};
window.startAmbientOnInteraction = function() {
window.initAmbient();
if (!ambientMuted && ambientAudio.paused) {
ambientAudio.play().catch(e => console.log("Playback failed/blocked:", e));
}
// Remove the listener once started successfully
document.removeEventListener("click", window.startAmbientOnInteraction);
document.removeEventListener("keydown", window.startAmbientOnInteraction);
};
window.toggleAmbient = function() {
window.initAmbient();
const btn = document.getElementById("ambient-toggle-btn");
if (!ambientAudio) return;
if (ambientMuted) {
// Turn ON
ambientMuted = false;
ambientAudio.volume = ambientVolume;
ambientAudio.play().catch(e => console.error(e));
if (btn) {
btn.innerText = "ON";
btn.style.background = "#39ff14";
btn.style.borderColor = "rgba(57, 255, 20, 0.4)";
btn.style.color = "#0a0e14";
btn.style.boxShadow = "0 0 10px rgba(57,255,20,0.4)";
}
} else {
// Turn OFF
ambientMuted = true;
ambientAudio.pause();
if (btn) {
btn.innerText = "OFF";
btn.style.background = "#ff4444";
btn.style.borderColor = "rgba(255, 68, 68, 0.4)";
btn.style.color = "#fff";
btn.style.boxShadow = "0 0 5px rgba(255,68,68,0.2)";
}
}
};
window.setAmbientVolume = function(val) {
ambientVolume = parseFloat(val);
const valSpan = document.getElementById("ambient-volume-val");
if (valSpan) {
valSpan.innerText = Math.round(ambientVolume * 100) + "%";
}
window.initAmbient();
if (ambientAudio) {
window.updateAudioVolumeActual();
}
};
let voicePlaying = false;
let voiceEnabled = true;
let lastSpokenText = "";
window.toggleVoice = function() {
voiceEnabled = !voiceEnabled;
const btn = document.getElementById("voice-toggle-btn");
if (voiceEnabled) {
if (btn) {
btn.innerText = "🔊 VOICE ON";
btn.style.background = "#39ff14";
btn.style.color = "#0a0e14";
btn.style.borderColor = "rgba(57, 255, 20, 0.4)";
btn.style.boxShadow = "0 0 10px rgba(57,255,20,0.4)";
}
} else {
if (window.speechSynthesis) window.speechSynthesis.cancel();
voicePlaying = false;
if (btn) {
btn.innerText = "🔇 VOICE OFF";
btn.style.background = "#2a3040";
btn.style.color = "#8a8070";
btn.style.borderColor = "rgba(138, 128, 112, 0.4)";
btn.style.boxShadow = "none";
}
}
};
window.testVoice = function() {
console.log("[AnalogTown] testVoice clicked");
if (!window.speechSynthesis) {
alert("Your browser does not support speechSynthesis.");
return;
}
voiceEnabled = true;
try { window.speechSynthesis.resume(); } catch (e) {}
const u = new SpeechSynthesisUtterance("Receiver online. This is a voice test from Analog Town.");
u.rate = 0.95;
u.pitch = 1.0;
u.volume = 1.0;
const voices = window.speechSynthesis.getVoices();
const enVoice = voices.find(v => v.lang && v.lang.toLowerCase().startsWith("en"));
if (enVoice) { u.voice = enVoice; console.log("[AnalogTown] test using voice:", enVoice.name, enVoice.lang); }
u.onstart = function() {
voicePlaying = true;
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#39ff14"; led.style.boxShadow = "0 0 8px #39ff14"; }
console.log("[AnalogTown] TEST utterance STARTED");
};
u.onend = function() {
voicePlaying = false;
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#2a3040"; led.style.boxShadow = "none"; }
console.log("[AnalogTown] TEST utterance ENDED");
};
u.onerror = function(e) {
console.error("[AnalogTown] TEST utterance ERROR:", e.error || e);
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#ff4444"; led.style.boxShadow = "0 0 8px #ff4444"; }
};
window.speechSynthesis.speak(u);
console.log("[AnalogTown] TEST speak() invoked, pending:", window.speechSynthesis.pending, "speaking:", window.speechSynthesis.speaking, "paused:", window.speechSynthesis.paused);
setTimeout(function() {
console.log("[AnalogTown] 1s after speak — speaking:", window.speechSynthesis.speaking, "paused:", window.speechSynthesis.paused);
if (window.speechSynthesis.paused) {
console.log("[AnalogTown] engine paused, attempting resume");
window.speechSynthesis.resume();
}
}, 1000);
};
const FEMALE_NAME_TOKENS = [
"sarah","chloe","priya","maple","mei","elena","martha","leah","mara",
"joan","lila","mira","idell","rusty","ronnie","tiffany","margaret",
"auntie","sister","granny","elder martha","aria","ronnie mae","mae",
"elder","mira okonkwo","lila mendes","an"
];
const MALE_NAME_TOKENS = [
"arthur","toby","jesse","buck","hollis","joaquin","daniel","wade","mason",
"tuan","kai","engineer daniyal","daniyal","pastor","preacher","ng","martin",
"captain arthur","captain","gerald","silas","alistair","marcus","corporal dale",
"eli","theo","pastor lin","councilman","architect"
];
const FEMALE_VOICE_POOL = [
"Samantha","Victoria","Karen","Moira","Allison","Susan","Ava","Fiona","Tessa",
"Veena","Kate","Serena","Google UK English Female","Google US English",
"Microsoft Zira","Microsoft Aria"
];
const MALE_VOICE_POOL = [
"Daniel","Alex","Fred","Tom","Aaron","Lee","Oliver","Rishi","Albert","Bruce",
"Junior","Ralph","Google UK English Male","Microsoft David","Microsoft Mark"
];
function _stableHash(s) {
let h = 0;
for (let i = 0; i < s.length; i++) {
h = ((h << 5) - h + s.charCodeAt(i)) | 0;
}
return Math.abs(h);
}
function _pickVoiceForAgent(name, ageRange) {
const voices = window.speechSynthesis.getVoices().filter(v => v.lang && v.lang.toLowerCase().startsWith("en"));
if (voices.length === 0) return { voice: null, pitch: 1.0, rate: 0.95 };
const lname = (name || "").toLowerCase();
let isFemale = FEMALE_NAME_TOKENS.some(t => lname.includes(t));
let isMale = MALE_NAME_TOKENS.some(t => lname.includes(t));
if (isFemale && isMale) {
if (lname.indexOf("captain arthur") >= 0 || lname.indexOf("arthur") >= 0) { isFemale = false; isMale = true; }
}
if (!isFemale && !isMale) {
const h = _stableHash(lname || "x");
isFemale = (h % 2) === 0;
}
const pool = isFemale ? FEMALE_VOICE_POOL : MALE_VOICE_POOL;
const hash = _stableHash(lname || "x");
const idx = hash % pool.length;
let voice = null;
for (let offset = 0; offset < pool.length; offset++) {
const candidate = pool[(idx + offset) % pool.length];
voice = voices.find(v => v.name === candidate);
if (voice) break;
voice = voices.find(v => v.name.toLowerCase().includes(candidate.toLowerCase()));
if (voice) break;
}
if (!voice) {
const filtered = voices.filter(v => isFemale ? /female|woman|samantha|victoria|karen|moira|tessa|aria/i.test(v.name) : /male|man|daniel|alex|fred|tom|david|mark/i.test(v.name));
voice = (filtered.length > 0 ? filtered[hash % filtered.length] : voices[hash % voices.length]);
}
const isElder = /\\b(6[0-9]|7[0-9]|8[0-9]|elder|granny|auntie|pastor|preacher|veteran|retired)\\b/i.test((ageRange || "") + " " + (name || ""));
const isYoung = /\\b(1[6-9]|2[0-4]|teen|teenager|young|apprentice)\\b/i.test((ageRange || "") + " " + (name || ""));
let pitch = 1.0 + ((hash % 9) - 4) * 0.03;
if (isElder) pitch -= 0.12;
if (isYoung) pitch += 0.08;
pitch = Math.max(0.6, Math.min(1.4, pitch));
const rate = 0.88 + (hash % 5) * 0.03;
return { voice: voice, pitch: pitch, rate: rate, isFemale: isFemale, isElder: isElder, isYoung: isYoung };
}
window.speakMonologue = function(text, name, ageRange) {
console.log("[AnalogTown] speakMonologue called:", { textLen: (text||"").length, name: name, voiceEnabled: voiceEnabled });
if (!voiceEnabled) { console.log("[AnalogTown] voice disabled, skipping"); return; }
if (!window.speechSynthesis || !window.SpeechSynthesisUtterance) {
console.warn("[AnalogTown] Browser SpeechSynthesis not available");
return;
}
if (!text) { console.log("[AnalogTown] empty text, skipping"); return; }
if (text === lastSpokenText) { console.log("[AnalogTown] same as last spoken, skipping"); return; }
lastSpokenText = text;
try { window.speechSynthesis.cancel(); } catch (e) {}
try { window.speechSynthesis.resume(); } catch (e) {}
const utter = new SpeechSynthesisUtterance(text);
utter.volume = 1.0;
const choice = _pickVoiceForAgent(name, ageRange);
if (choice.voice) {
utter.voice = choice.voice;
console.log("[AnalogTown] voice for", name, "→", choice.voice.name, "(female=" + choice.isFemale + ", elder=" + choice.isElder + ", young=" + choice.isYoung + ")");
}
utter.pitch = choice.pitch;
utter.rate = choice.rate;
console.log("[AnalogTown] pitch=" + utter.pitch.toFixed(2) + " rate=" + utter.rate.toFixed(2));
utter.onstart = function() {
voicePlaying = true;
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#39ff14"; led.style.boxShadow = "0 0 8px #39ff14"; }
console.log("[AnalogTown] utterance started");
};
utter.onend = function() {
voicePlaying = false;
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#2a3040"; led.style.boxShadow = "none"; }
console.log("[AnalogTown] utterance ended");
};
utter.onerror = function(e) {
voicePlaying = false;
const led = document.getElementById("voice-status-led");
if (led) { led.style.background = "#ff4444"; led.style.boxShadow = "0 0 8px #ff4444"; }
console.error("[AnalogTown] utterance error:", e);
};
setTimeout(function() {
window.speechSynthesis.speak(utter);
console.log("[AnalogTown] speak() called");
}, 80);
};
if (window.speechSynthesis) {
window.speechSynthesis.getVoices();
window.speechSynthesis.onvoiceschanged = function() {
const v = window.speechSynthesis.getVoices();
console.log("[AnalogTown] voices loaded:", v.length);
};
}
window.updateAudioVolumeActual = function() {
if (!ambientAudio) return;
if (ambientMuted) {
ambientAudio.volume = 0;
return;
}
// Calculate target volume based on active signal bars
const activeBars = document.querySelectorAll(".signal-bar.active").length;
const signalFade = 1.0 - (activeBars / 5.0);
const baseTarget = ambientVolume * signalFade;
// Duck to 15% of user ambient volume while voice is playing
targetActualVolume = voicePlaying ? ambientVolume * 0.15 : baseTarget;
// Smoothly interpolate currentActualVolume towards targetActualVolume
const diff = targetActualVolume - currentActualVolume;
if (Math.abs(diff) > 0.01) {
currentActualVolume += diff * 0.15;
} else {
currentActualVolume = targetActualVolume;
}
// Set audio volume
ambientAudio.volume = Math.max(0, Math.min(1, currentActualVolume));
};
// Listeners for initial interaction
document.addEventListener("click", window.startAmbientOnInteraction);
document.addEventListener("keydown", window.startAmbientOnInteraction);
// Polling interval for smooth fade
setInterval(window.updateAudioVolumeActual, 100);
let _voiceContainerWarned = false;
let _voicePollTick = 0;
setInterval(function() {
_voicePollTick++;
let c = document.getElementById("voice-player-container");
if (!c) {
c = document.querySelector("[data-text][data-name]");
}
if (!c) {
if (!_voiceContainerWarned && _voicePollTick % 25 === 0) {
console.warn("[AnalogTown] voice-player-container NOT FOUND in DOM after", _voicePollTick * 0.4, "s — sprite clicks won't trigger TTS until it appears");
_voiceContainerWarned = true;
}
return;
}
if (_voiceContainerWarned) {
console.log("[AnalogTown] voice-player-container found:", c);
_voiceContainerWarned = false;
}
const text = c.getAttribute("data-text") || "";
const name = c.getAttribute("data-name") || "";
const age = c.getAttribute("data-age") || "";
if (text && text !== lastSpokenText) {
console.log("[AnalogTown] poll detected new data-text:", text.slice(0, 60), "...");
window.speakMonologue(text, name, age);
}
}, 400);
// Chrome workaround: speech engine auto-pauses after ~15s of inactivity. Keep it warm.
setInterval(function() {
if (window.speechSynthesis && window.speechSynthesis.speaking && window.speechSynthesis.paused) {
window.speechSynthesis.resume();
}
}, 5000);
window.selectTimelineDay = function(n) {
console.log("selectTimelineDay called with day:", n);
const container = document.getElementById("selected-day");
if (container) {
const input = container.querySelector("input") || container.querySelector("textarea");
if (input) {
let prototype = Object.getPrototypeOf(input);
let descriptor = Object.getOwnPropertyDescriptor(prototype, "value");
while (prototype && !descriptor) {
prototype = Object.getPrototypeOf(prototype);
descriptor = Object.getOwnPropertyDescriptor(prototype, "value");
}
const setter = descriptor ? descriptor.set : null;
if (setter) {
setter.call(input, String(n));
} else {
input.value = String(n);
}
input.dispatchEvent(new Event("input", { bubbles: true }));
setTimeout(() => {
const btn = document.getElementById("select-day-trigger");
if (btn) {
console.log("Clicking select-day-trigger button");
btn.click();
} else {
console.error("select-day-trigger button not found");
}
}, 50);
} else {
console.error("Input/textarea not found inside #selected-day");
}
} else {
console.error("#selected-day container not found");
}
};
"""
AUDIO_PATH = Path(__file__).parent / "static_ambient.wav"
AUDIO_URL = f"/gradio_api/file={AUDIO_PATH.resolve()}"
CUSTOM_JS = CUSTOM_JS_TEMPLATE.replace("__AUDIO_URL__", AUDIO_URL)
def create_app():
"""Build the Gradio application."""
available_towns = get_available_towns()
with gr.Blocks(
title="Analog Town — Shortwave Social Simulator",
) as app:
# ── State ──
town_state = gr.State("")
sim_result_state = gr.State("")
timeline_state = gr.State("")
# ── Hidden Map Communication Inputs ──
selected_agent_id = gr.Textbox(elem_id="selected-agent-id", elem_classes=["hidden-component"], visible=True)
select_agent_trigger = gr.Button("Select Agent Trigger", elem_id="select-agent-trigger", elem_classes=["hidden-component"], visible=True)
# ── Hidden Timeline Day Bridge ──
selected_day = gr.Textbox(elem_id="selected-day", elem_classes=["hidden-component"], visible=True)
select_day_trigger = gr.Button("Select Day Trigger", elem_id="select-day-trigger", elem_classes=["hidden-component"], visible=True)
# ── Header ──
with gr.Row(elem_id="header-block"):
with gr.Column():
gr.HTML("""
""")
# ── Main Layout ──
# === TOP CONTROL BAR (town + broadcast + transmit, all in one strip) ===
with gr.Group(elem_id="control-bar"):
with gr.Row(equal_height=True):
with gr.Column(scale=2, min_width=200):
gr.HTML('📡 TOWN
')
town_dropdown = gr.Dropdown(
choices=available_towns,
value=available_towns[0] if available_towns else None,
show_label=False,
container=False,
interactive=True,
elem_id="town-selector",
)
load_btn = gr.Button(
"⚡ Load Town",
size="sm",
elem_classes=["export-btn"],
)
with gr.Column(scale=5, min_width=320):
gr.HTML('📢 BROADCAST
')
event_title = gr.Textbox(
placeholder="Broadcast title (e.g., Station Hotel Development)",
show_label=False,
container=False,
lines=1,
elem_id="event-title",
)
event_content = gr.Textbox(
placeholder="Enter the news, proposal, or story event to broadcast into the town...",
show_label=False,
container=False,
lines=2,
elem_id="event-content",
)
with gr.Column(scale=2, min_width=180):
gr.HTML('▸ TRANSMIT
')
with gr.Row(equal_height=True):
transmit_btn = gr.Button(
"🔊 TRANSMIT",
variant="primary",
elem_classes=["transmit-btn"],
elem_id="transmit-btn",
scale=3,
)
reset_timeline_btn = gr.Button(
"🔁 RESET",
size="sm",
elem_classes=["export-btn"],
elem_id="reset-timeline-btn",
scale=1,
)
sim_status = gr.Textbox(
show_label=False,
container=False,
lines=2,
max_lines=2,
interactive=False,
placeholder="Awaiting broadcast...",
elem_id="sim-status",
)
# === TIMELINE STRIP (below control bar, full width) ===
timeline_strip = gr.HTML(
value=format_timeline_strip(""),
elem_id="timeline-strip-container",
)
# === CREATIVE MODE ACCORDION ===
with gr.Accordion("🎨 CREATIVE MODE — Design Your Own Town", open=False, elem_id="creative-mode"):
with gr.Row():
with gr.Column(scale=4):
concept_input = gr.Textbox(
label="🎨 TOWN CONCEPT",
placeholder="e.g., a haunted seaside lighthouse where the keeper has gone missing and the village is split on whether to investigate or move on",
lines=4,
elem_id="creative-concept",
)
with gr.Column(scale=2):
agent_count_slider = gr.Slider(
label="# AGENTS",
minimum=3,
maximum=6,
step=1,
value=4,
elem_id="creative-agent-count",
)
map_upload = gr.File(
label="📁 UPLOAD MAP PNG (optional)",
file_types=["image"],
type="filepath",
elem_id="creative-map-upload",
)
with gr.Row():
creative_generate_btn = gr.Button(
"🪄 GENERATE WITH AI",
variant="secondary",
elem_classes=["export-btn"],
elem_id="creative-generate-btn",
)
creative_save_btn = gr.Button(
"💾 SAVE & LOAD",
variant="primary",
elem_classes=["transmit-btn"],
elem_id="creative-save-btn",
)
town_json_code = gr.Code(
label="EDITABLE TOWN JSON",
language="json",
lines=14,
elem_id="creative-json",
)
creative_status = gr.Markdown("_Status: idle_", elem_id="creative-status")
# === MAIN STAGE (map+dossier on left | receiver on right) ===
with gr.Row(elem_id="main-stage", equal_height=True):
# ── LEFT: Town Map + Agent Dossier ──
with gr.Column(scale=6, min_width=420, elem_id="map-panel"):
gr.HTML('')
day_badge = gr.HTML(
value="",
elem_id="day-badge-container",
)
town_map = gr.HTML(
value="Load a town to see the map
",
elem_id="town-map",
)
with gr.Accordion("👤 AGENT DOSSIER", open=True, elem_id="dossier-accordion"):
agent_profile = gr.HTML(
value="Select an agent on the map to inspect dossier
",
elem_id="agent-profile",
)
# ── RIGHT: Shortwave Receiver ──
with gr.Column(scale=5, min_width=380, elem_id="receiver-panel"):
gr.HTML('')
freq_display = gr.HTML(
value=format_frequency_display(88.0, None, 0),
elem_id="freq-display",
)
freq_slider = gr.Slider(
minimum=87.0,
maximum=108.0,
value=88.0,
step=0.1,
label="◄ TUNE FREQUENCY ►",
elem_id="freq-slider",
)
monologue_display = gr.HTML(
value=format_monologue(None, None),
elem_id="monologue",
)
voice_player = gr.HTML(
value=_build_voice_player(None),
elem_id="voice-player",
elem_classes=["hidden-component"],
)
with gr.Row(equal_height=True):
with gr.Column(scale=1, min_width=160):
gr.HTML('')
emotion_display = gr.HTML(
value=format_emotion_bars(None),
elem_id="emotions",
)
with gr.Column(scale=1, min_width=160):
gr.HTML('')
action_display = gr.HTML(
value=format_action_display(None),
elem_id="actions",
)
ambient_control = gr.HTML(
value=f"""
""",
elem_id="ambient-control",
)
# ── BOTTOM PANEL: Trace Inspector & Export (collapsed by default) ──
with gr.Row(elem_id="bottom-stage"):
with gr.Column():
with gr.Accordion("🔍 SIGNAL TRACE LOG", open=False):
trace_json = gr.Code(
label="Raw Transition JSON",
language="json",
lines=15,
elem_id="trace-json",
)
with gr.Accordion("📦 EXPORT & SHARE", open=False):
with gr.Row():
with gr.Column():
export_local_btn = gr.Button(
"📁 Export to Local ZIP",
elem_classes=["export-btn"],
)
export_file = gr.File(
label="Download Export",
interactive=False,
)
with gr.Column():
hub_repo_id = gr.Textbox(
label="HF HUB REPO ID",
placeholder="username/analog-town-traces",
)
export_hub_btn = gr.Button(
"🤗 Upload to HF Hub",
elem_classes=["export-btn"],
)
export_status = gr.Textbox(
label="EXPORT STATUS",
lines=2,
interactive=False,
)
# ── Safety Footer ──
gr.HTML(f"""
""")
# ── Event Handlers ──
_load_outputs = [
town_map, agent_profile, town_state, event_title, event_content,
sim_result_state, timeline_state, timeline_strip, day_badge,
sim_status,
freq_display, monologue_display, emotion_display, action_display, trace_json, voice_player,
freq_slider,
]
load_btn.click(
fn=on_load_town,
inputs=[town_dropdown],
outputs=_load_outputs,
)
town_dropdown.change(
fn=on_load_town,
inputs=[town_dropdown],
outputs=_load_outputs,
)
creative_generate_btn.click(
fn=on_generate_creative_town,
inputs=[concept_input, agent_count_slider, town_json_code],
outputs=[town_json_code, creative_status],
)
creative_save_btn.click(
fn=on_save_load_creative_town,
inputs=[town_json_code, map_upload],
outputs=_load_outputs + [town_dropdown, creative_status],
)
# Map Click Event Selector
select_agent_trigger.click(
fn=on_select_agent,
inputs=[selected_agent_id, town_state, sim_result_state],
outputs=[
town_map,
agent_profile,
freq_slider,
freq_display,
monologue_display,
emotion_display,
action_display,
trace_json,
voice_player,
],
)
# Transmit
transmit_btn.click(
fn=on_transmit,
inputs=[town_state, event_title, event_content, timeline_state],
outputs=[sim_status, sim_result_state, town_map, timeline_state, timeline_strip, day_badge],
show_progress_on=[sim_status],
)
# Reset timeline
reset_timeline_btn.click(
fn=on_reset_timeline,
inputs=[town_state],
outputs=[sim_result_state, town_map, timeline_strip, day_badge],
)
# Timeline day selection bridge
select_day_trigger.click(
fn=on_select_timeline_day,
inputs=[selected_day, town_state, timeline_state],
outputs=[
sim_result_state,
town_map,
freq_display,
monologue_display,
emotion_display,
action_display,
trace_json,
voice_player,
timeline_strip,
day_badge,
],
)
# Tune frequency
freq_slider.change(
fn=tune_frequency,
inputs=[freq_slider, sim_result_state],
outputs=[freq_display, monologue_display, emotion_display, action_display, trace_json, voice_player],
)
# Export local
export_local_btn.click(
fn=on_export_local,
inputs=[sim_result_state, town_state],
outputs=[export_file, export_status],
)
# Export to Hub
export_hub_btn.click(
fn=on_export_hub,
inputs=[sim_result_state, town_state, hub_repo_id],
outputs=[export_status],
)
app.load(
fn=lambda: on_load_town(available_towns[0] if available_towns else None),
outputs=_load_outputs,
)
return app
# ──────────────────────────────────────────────
# Main
# ──────────────────────────────────────────────
if __name__ == "__main__":
app = create_app()
app.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True,
allowed_paths=[str(Path(__file__).parent)], # Allow serving local images
css=CUSTOM_CSS,
js=CUSTOM_JS,
theme=gr.themes.Base(
primary_hue=gr.themes.colors.amber,
secondary_hue=gr.themes.colors.green,
neutral_hue=gr.themes.colors.gray,
font=gr.themes.GoogleFont("IBM Plex Mono"),
),
)