"""File management callbacks.
This module handles:
- Loading map data when a map is selected from the Workspace table
- New map modal open/close/create
- Save and export functionality
- Status updates
Authoring + Export Workflow
---------------------------
The mapper uses SQLite as the authoring source of truth and exports
JSON zone files for the game server:
- **SQLite maps**: Authoring source with coordinates
- **Zone files** (``data/zones/*.json``): Game truth without coordinates
Authors work with SQLite maps. Zone files are exported for game server use.
Component Dependencies
----------------------
**Inputs:**
- ``initial-load``: Interval trigger for startup
- ``zones-files-store``: List of exported zone files
- ``selected-file``: Currently selected map ID
- ``workspace-map-row`` (pattern): Clickable Workspace table rows
- ``new-map-btn``: Open new map modal
- ``new-map-cancel-btn``: Close modal
- ``new-map-create-btn``: Create new zone
- ``save-map-btn``: Save current map
- ``export-zone-btn``: Export zone JSON
**Outputs:**
- ``zone-files-list-container``: Rendered zone export list
- ``exports-status-indicator``: Export status display
- ``selected-file``: Selected map ID
- ``current-zone-data``: Loaded map data
- ``current-zone``: Zone name display
- ``selected-room``: Room selection (cleared on map change)
- ``new-map-modal``: Modal visibility
- ``has-unsaved-changes``: Unsaved flag
- ``save-map-btn``: Save button state
- ``status-indicator``: Status display
"""
import json
import re
import time
from pathlib import Path
from typing import Any
import dash_bootstrap_components as dbc
from dash import ALL, Input, Output, State, callback, ctx, html, no_update
from pipeworks_mud_mapper.services import map_db_service, zone_service
from pipeworks_mud_mapper.services.app_config import get_path_settings
from pipeworks_mud_mapper.services.io_queue import (
forget_io_job,
get_io_job_status,
submit_io_job,
)
from pipeworks_mud_mapper.services.state import ZoneAction, apply_zone_action
# Paths for SQLite + export workflow (user-configurable via config/server.ini)
PATHS = get_path_settings()
DB_PATH = PATHS["db_path"]
ZONES_DIR = PATHS["zones_dir"]
# =============================================================================
# File Listing Cache
# =============================================================================
# These cache structures reduce repeated filesystem scans when callbacks are
# triggered in quick succession.
_FILE_LIST_CACHE: dict[Path, tuple[float, list[Path]]] = {}
def _room_feedback_payload(content: Any) -> dict[str, Any]:
"""Build a timestamped payload for room form feedback."""
return {"content": content, "ts": time.monotonic()}
def _save_map_job(map_file: Any, db_path: Path) -> None:
"""Persist a map in a background thread.
Revision bumps happen in the UI callback so the in-memory state stays
consistent with what gets written to SQLite.
"""
map_db_service.save_map(map_file, db_path=db_path)
def _export_zone_job(
export_map: Any,
export_path: Path,
db_path: Path,
updated_map: Any,
) -> None:
"""Export a zone file and persist updated map metadata in the background.
The export uses the pre-bump map metadata for provenance. After the zone
is written, the updated map (with incremented map_version) is saved back
to SQLite without touching map_revision.
"""
zone_service.export_zone(export_map, export_path)
map_db_service.save_map(updated_map, db_path=db_path)
# =============================================================================
# Export List Callbacks
# =============================================================================
[docs]
@callback(
Output("zones-files-store", "data"),
Input("initial-load", "n_intervals"),
Input("room-feedback-export", "data"),
prevent_initial_call=False,
)
def load_zone_files_list(_: int, __: dict | None) -> list[str]:
"""Load list of exported zone files from the zones directory.
This callback is triggered on initial page load and after export
feedback updates so the list reflects newly exported files.
"""
# Ensure the export directory exists so the UI list can render consistently.
ZONES_DIR.mkdir(parents=True, exist_ok=True)
# Zone files are game-truth JSON (no coordinates), so we list *.json.
files = zone_service.list_zone_files(ZONES_DIR)
return [f.name for f in files]
[docs]
@callback(
Output("zone-files-list-container", "children"),
Input("zones-files-store", "data"),
)
def render_zone_files_list(files: list[str]) -> list:
"""Render the exported zone file list."""
# Zones are display-only so users can see what has been exported.
if not files:
return [html.Span("No zone exports found", className="text-muted fst-italic")]
items = []
for filename in files:
icon_class = "bi bi-file-earmark-code me-2"
display_name = filename
if filename.endswith(".json"):
display_name = filename[:-5]
items.append(
html.Div(
[
html.I(className=icon_class),
html.Span(display_name),
],
id={"type": "zone-file-item", "filename": filename},
className="mb-1 p-1 rounded file-item",
style={"cursor": "pointer"},
n_clicks=0,
)
)
return items
[docs]
@callback(
Output("exports-status-indicator", "children"),
Input("room-feedback-export", "data"),
)
def render_export_status(payload: dict | None) -> Any:
"""Render the latest export status feedback in the exports card."""
if not isinstance(payload, dict):
return no_update
content = payload.get("content")
if content is None:
return no_update
return content
[docs]
@callback(
Output("zone-json-modal", "is_open"),
Output("zone-json-modal-title", "children"),
Output("zone-json-modal-body", "children"),
Output("selected-zone-file", "data", allow_duplicate=True),
Input({"type": "zone-file-item", "filename": ALL}, "n_clicks"),
Input("zone-json-close-btn", "n_clicks"),
prevent_initial_call=True,
)
def handle_zone_file_click(zone_clicks: list[int], close_clicks: int | None) -> tuple:
"""Open a modal showing the selected zone JSON."""
trigger = ctx.triggered_id
if trigger == "zone-json-close-btn":
return False, no_update, no_update, no_update
if not any(zone_clicks):
return no_update, no_update, no_update, no_update
if not trigger or not isinstance(trigger, dict):
return no_update, no_update, no_update, no_update
filename = trigger.get("filename")
if not filename:
return no_update, no_update, no_update, no_update
file_path = ZONES_DIR / filename
if not file_path.exists():
feedback = dbc.Alert(
f"Zone file not found: {filename}",
color="warning",
className="mb-0",
)
return True, "Zone JSON", feedback, filename
try:
data = json.loads(file_path.read_text(encoding="utf-8"))
pretty = json.dumps(data, indent=2, sort_keys=True)
content = html.Pre(pretty, className="mb-0 small")
return True, f"Zone JSON: {filename}", content, filename
except json.JSONDecodeError as exc:
feedback = dbc.Alert(
f"Invalid JSON in {filename}: {exc}",
color="danger",
className="mb-0",
)
return True, "Zone JSON", feedback, filename
[docs]
@callback(
Output("file-properties-name", "children"),
Output("file-properties-type", "children"),
Output("file-properties-delete-btn", "disabled"),
Input("selected-file", "data"),
Input("selected-zone-file", "data"),
Input("selected-room", "data"),
Input("current-zone-data", "data"),
)
def render_file_properties(
selected_file: str | None,
selected_zone_file: str | None,
selected_room: str | None,
zone_data: dict | None,
) -> tuple:
"""Render the file properties summary in the right column."""
if selected_room and zone_data:
rooms = zone_data.get("rooms", {})
room = rooms.get(selected_room)
if room:
coords = room.get("coords", [0, 0, 0])
name = html.Span(selected_room)
detail = html.Div(
[
dbc.Badge("Room", color="secondary", className="me-2"),
html.Span(room.get("name", "")),
html.Div(
f"Coords: {coords[0]}, {coords[1]}, {coords[2]}",
className="text-muted small",
),
]
)
return name, detail, True
if selected_file:
name = html.Span(selected_file)
badge = dbc.Badge("Map", color="primary", className="me-2")
return name, html.Div([badge, html.Span("Selected")]), False
if selected_zone_file:
name = html.Span(selected_zone_file)
badge = dbc.Badge("Zone export", color="info", className="me-2")
return name, html.Div([badge, html.Span("Selected")]), False
return html.Span("No file selected", className="text-muted"), "", True
[docs]
@callback(
Output("file-delete-confirm-modal", "is_open"),
Output("file-delete-confirm-body", "children"),
Output("file-delete-pending", "data"),
Input("file-properties-delete-btn", "n_clicks"),
Input("file-delete-cancel-btn", "n_clicks"),
State("selected-file", "data"),
State("selected-zone-file", "data"),
prevent_initial_call=True,
)
def request_file_delete(
delete_clicks: int | None,
cancel_clicks: int | None,
selected_file: str | None,
selected_zone_file: str | None,
) -> tuple:
"""Open confirmation modal when a delete button is clicked."""
trigger = ctx.triggered_id
if trigger == "file-delete-cancel-btn":
return False, no_update, None
if not delete_clicks:
return no_update, no_update, no_update
# Prefer the currently loaded map over a zone export selection.
if selected_file:
filename = selected_file
delete_type = "file-delete-btn"
label = "map"
badge_color = "primary"
path_hint = DB_PATH
elif selected_zone_file:
filename = selected_zone_file
delete_type = "zone-file-delete-btn"
label = "zone export"
badge_color = "info"
path_hint = ZONES_DIR / filename
else:
return no_update, no_update, no_update
body = html.Div(
[
html.P("Are you sure you want to delete this file?"),
html.Div(
[
dbc.Badge(label, color=badge_color, className="me-2"),
html.Span(filename, className="fw-bold"),
],
className="mb-1",
),
html.Div(
[
html.Span("Path: ", className="text-muted"),
html.Code(str(path_hint)),
],
className="small text-muted",
),
],
className="mb-0",
)
return True, body, {"type": delete_type, "filename": filename}
[docs]
@callback(
Output("zones-files-store", "data", allow_duplicate=True),
Output("selected-file", "data", allow_duplicate=True),
Output("current-zone-data", "data", allow_duplicate=True),
Output("has-unsaved-changes", "data", allow_duplicate=True),
Output("file-delete-confirm-modal", "is_open", allow_duplicate=True),
Input("file-delete-confirm-btn", "n_clicks"),
State("file-delete-pending", "data"),
State("selected-file", "data"),
prevent_initial_call=True,
)
def confirm_file_delete(
confirm_clicks: int | None,
pending: dict | None,
selected_file: str | None,
) -> tuple:
"""Delete a file after confirmation and refresh the relevant list."""
if not confirm_clicks or not pending:
return no_update, no_update, no_update, no_update, no_update
delete_type = pending.get("type")
filename = pending.get("filename")
if not delete_type or not filename:
return no_update, no_update, no_update, no_update, no_update
zones_update = no_update
selected_update = no_update
zone_data_update = no_update
unsaved_update = no_update
if delete_type == "file-delete-btn":
# Remove the map from SQLite; no filesystem delete needed.
map_db_service.delete_map(filename, db_path=DB_PATH)
if filename == selected_file:
selected_update = None
zone_data_update = None
unsaved_update = False
else:
file_path = ZONES_DIR / filename
if file_path.exists():
file_path.unlink()
files = zone_service.list_zone_files(ZONES_DIR)
_FILE_LIST_CACHE[ZONES_DIR] = (time.monotonic(), files)
zones_update = [f.name for f in files]
return (
zones_update,
selected_update,
zone_data_update,
unsaved_update,
False,
)
[docs]
@callback(
Output("selected-file", "data"),
Output("current-zone-data", "data"),
Output("current-zone", "children"),
Output("has-unsaved-changes", "data", allow_duplicate=True),
Output("selected-zone-file", "data", allow_duplicate=True),
Output("selected-room", "data", allow_duplicate=True),
Input({"type": "workspace-map-row", "map_id": ALL}, "n_clicks"),
State("selected-file", "data"),
prevent_initial_call=True,
)
def handle_file_click(
map_clicks: list[int],
current_file: str | None,
) -> tuple:
"""Load map data when a Workspace table row is clicked.
Parameters
----------
map_clicks : list[int]
Click counts for workspace map rows.
current_file : str | None
Currently selected map ID (used to avoid redundant reloads).
Returns
-------
tuple
(selected_file, zone_data, zone_display, has_unsaved) or no_update tuple.
"""
# Bail out early when nothing has been clicked in either list.
if not any(map_clicks):
print("[DEBUG] handle_file_click: no clicks, returning no_update")
return no_update, no_update, no_update, no_update, no_update, no_update
# Log the trigger details so we can trace clicks across both lists.
print("[DEBUG] handle_file_click: " f"map_clicks={map_clicks}, current={current_file}")
print(f"[DEBUG] handle_file_click: triggered_id={ctx.triggered_id}")
# The triggered_id includes the pattern-matching payload with filename/type.
triggered = ctx.triggered_id
if not triggered or not isinstance(triggered, dict):
print("[DEBUG] handle_file_click: no valid trigger, returning no_update")
return no_update, no_update, no_update, no_update, no_update, no_update
map_id = triggered.get("map_id")
if not map_id:
print("[DEBUG] handle_file_click: no filename in trigger, returning no_update")
return no_update, no_update, no_update, no_update, no_update, no_update
# Avoid reloading the same file if it is already selected.
if map_id == current_file:
print(f"[DEBUG] handle_file_click: same map {map_id}, returning no_update")
return no_update, no_update, no_update, no_update, no_update, no_update
# Load the selected map file and reset unsaved changes.
action = ZoneAction(type="LOAD_MAP", payload={"map_id": map_id})
transition = apply_zone_action(None, action)
if not transition.changed or transition.zone_data is None:
print(f"Error loading map {map_id}")
return no_update, no_update, no_update, no_update, no_update, no_update
zone_name = transition.effects.get("zone_name", map_id)
return map_id, transition.zone_data, f"Zone: {zone_name}", False, None, None
# =============================================================================
# New Map Modal Callbacks
# =============================================================================
[docs]
@callback(
Output("new-map-modal", "is_open"),
Output("new-map-feedback", "children"),
Output("new-zone-id", "value"),
Output("new-zone-name", "value"),
Output("new-zone-description", "value"),
Input("new-map-btn", "n_clicks"),
Input("new-map-cancel-btn", "n_clicks"),
Input("new-map-create-btn", "n_clicks"),
State("new-zone-id", "value"),
State("new-zone-name", "value"),
State("new-zone-description", "value"),
prevent_initial_call=True,
)
def handle_new_map_modal(
open_clicks: int,
cancel_clicks: int,
create_clicks: int,
zone_id: str,
zone_name: str,
description: str,
) -> tuple:
"""Open, close, and create new maps from a single modal callback.
This consolidates the previous open/close/create callbacks so only
one callback owns the modal state. It routes behavior based on the
triggering input and only runs creation logic for the Create button.
"""
trigger = ctx.triggered_id
# Open the modal when the "New Map" button is clicked.
if trigger == "new-map-btn":
return True, no_update, no_update, no_update, no_update
# Close the modal when Cancel is clicked.
if trigger == "new-map-cancel-btn":
return False, no_update, no_update, no_update, no_update
# Only the Create button should run creation logic.
if trigger != "new-map-create-btn" or not create_clicks:
return no_update, no_update, no_update, no_update, no_update
# Normalize inputs to avoid whitespace and casing issues.
zone_id = (zone_id or "").strip().lower()
zone_name = (zone_name or "").strip()
description = (description or "").strip()
# Validate zone_id.
if not zone_id:
feedback = dbc.Alert("Zone ID is required.", color="danger", className="mb-0")
return True, feedback, no_update, no_update, no_update
if not re.match(r"^[a-z][a-z0-9_]*$", zone_id):
feedback = dbc.Alert(
"Zone ID must start with a letter and contain only "
"lowercase letters, numbers, and underscores.",
color="danger",
className="mb-0",
)
return True, feedback, no_update, no_update, no_update
# Validate zone_name.
if not zone_name:
feedback = dbc.Alert("Zone Name is required.", color="danger", className="mb-0")
return True, feedback, no_update, no_update, no_update
# Check if map already exists in SQLite.
if map_db_service.map_exists(zone_id, db_path=DB_PATH):
feedback = dbc.Alert(
f"A map with ID '{zone_id}' already exists.",
color="warning",
className="mb-0",
)
return True, feedback, no_update, no_update, no_update
# Create and save the map in SQLite using the MapFile model.
map_file = zone_service.create_new_map_file(
zone_id=zone_id,
name=zone_name,
spawn_room_name="Spawn Room",
description=description,
)
map_db_service.save_map(map_file, db_path=DB_PATH)
# Close modal and clear form on success.
return False, "", "", "", ""
# =============================================================================
# Save/Export/Status Callbacks
# =============================================================================
[docs]
@callback(
Output("save-map-btn", "disabled"),
Output("export-zone-btn", "disabled"),
Output("status-indicator", "children"),
Input("has-unsaved-changes", "data"),
Input("selected-file", "data"),
)
def update_save_status(has_unsaved: bool, selected_file: str | None) -> tuple:
"""Update save/export button state and status indicator.
Shows appropriate status based on current state:
- No map loaded: disabled buttons
- Unsaved changes: enabled save, disabled export
- All saved: disabled save, enabled export
Parameters
----------
has_unsaved : bool
Whether there are unsaved changes.
selected_file : str | None
Currently selected map ID.
Returns
-------
tuple
(save_disabled, export_disabled, status_text).
"""
print(f"[DEBUG] update_save_status: has_unsaved={has_unsaved}, file={selected_file}")
if not selected_file:
print("[DEBUG] update_save_status: no map loaded")
return True, True, html.Span("No map loaded", className="text-muted")
# Display full map ID for clarity
display_name = selected_file
if has_unsaved:
print("[DEBUG] update_save_status: unsaved changes - save=ENABLED")
return False, True, html.Span(f"Unsaved: {display_name}", className="text-muted")
print("[DEBUG] update_save_status: saved - export=ENABLED")
saved_alert = dbc.Alert(
f"Saved: {display_name}",
color="success",
className="mb-0 py-1",
duration=3000,
dismissable=True,
fade=True,
)
return True, False, saved_alert
[docs]
@callback(
Output("has-unsaved-changes", "data", allow_duplicate=True),
Output("room-feedback-save", "data"),
Output("io-jobs", "data", allow_duplicate=True),
Output("current-zone-data", "data", allow_duplicate=True),
Input("save-map-btn", "n_clicks"),
State("current-zone-data", "data"),
State("selected-file", "data"),
State("io-jobs", "data"),
prevent_initial_call=True,
)
def save_map_to_file(
n_clicks: int,
zone_data: dict | None,
selected_file: str | None,
io_jobs: dict | None,
) -> tuple:
"""Save the current map data to SQLite.
Parameters
----------
n_clicks : int
Click count for Save button.
zone_data : dict | None
Current map data to save.
selected_file : str | None
Target map ID.
Returns
-------
tuple
(unsaved_flag, feedback_alert, jobs, updated_zone_data).
On success: False and success message.
On error: no_update and error message.
"""
if not n_clicks or not zone_data or not selected_file:
return no_update, no_update, no_update, no_update
try:
# Convert dict to MapFile and save
from pipeworks_mud_mapper.models import MapFile
map_file = MapFile.from_dict(zone_data)
display_name = selected_file
# Increment revision on every explicit save to track authoring history.
map_file.bump_revision()
updated_zone_data = map_file.to_dict_with_list_coords()
job_id = submit_io_job(_save_map_job, map_file, DB_PATH)
jobs = list((io_jobs or {}).get("jobs", []))
jobs.append(
{
"id": job_id,
"type": "save",
"display_name": display_name,
}
)
feedback = dbc.Alert(
f"Saving: {display_name}",
color="info",
className="mb-0 py-2",
duration=3000,
)
return True, _room_feedback_payload(feedback), {"jobs": jobs}, updated_zone_data
except Exception as e:
feedback = dbc.Alert(
f"Error saving: {e}",
color="danger",
className="mb-0 py-2",
)
return no_update, _room_feedback_payload(feedback), no_update, no_update
[docs]
@callback(
Output("room-feedback-export", "data"),
Output("io-jobs", "data", allow_duplicate=True),
Output("current-zone-data", "data", allow_duplicate=True),
Input("export-zone-btn", "n_clicks"),
State("current-zone-data", "data"),
State("selected-file", "data"),
State("io-jobs", "data"),
prevent_initial_call=True,
)
def export_zone_to_file(
n_clicks: int,
zone_data: dict | None,
selected_file: str | None,
io_jobs: dict | None,
) -> Any:
"""Export the current map as a zone file (strips coordinates).
Exports to data/zones/{name}.json, creating the game truth file
that the MUD server consumes.
Parameters
----------
n_clicks : int
Click count for Export button.
zone_data : dict | None
Current map data to export.
selected_file : str | None
Source map ID (used to derive export name).
Returns
-------
str
Feedback alert component.
"""
if not n_clicks or not zone_data or not selected_file:
return no_update, no_update, no_update
# Derive export path from map ID.
export_path = ZONES_DIR / f"{selected_file}.json"
try:
# Ensure zones directory exists
ZONES_DIR.mkdir(parents=True, exist_ok=True)
# Convert dict to MapFile and export
from pipeworks_mud_mapper.models import MapFile
map_file = MapFile.from_dict(zone_data)
export_map = map_file.model_copy(deep=True)
updated_map = map_file.model_copy(deep=True)
# Increment map_version on export and persist back to the map file.
updated_map.bump_version()
job_id = submit_io_job(_export_zone_job, export_map, export_path, DB_PATH, updated_map)
jobs = list((io_jobs or {}).get("jobs", []))
jobs.append(
{
"id": job_id,
"type": "export",
"display_name": export_path.stem,
}
)
feedback = dbc.Alert(
f"Export queued: {export_path.name} (coordinates stripped)",
color="info",
className="mb-0 py-2",
duration=4000,
)
updated_zone_data = updated_map.to_dict_with_list_coords()
return _room_feedback_payload(feedback), {"jobs": jobs}, updated_zone_data
except Exception as e:
feedback = dbc.Alert(
f"Error exporting: {e}",
color="danger",
className="mb-0 py-2",
)
return _room_feedback_payload(feedback), no_update, no_update
[docs]
@callback(
Output("io-jobs", "data"),
Output("room-feedback-save", "data", allow_duplicate=True),
Output("room-feedback-export", "data", allow_duplicate=True),
Output("has-unsaved-changes", "data", allow_duplicate=True),
Input("io-job-poll", "n_intervals"),
State("io-jobs", "data"),
prevent_initial_call="initial_duplicate",
)
def poll_io_jobs(n_intervals: int, io_jobs: dict | None) -> tuple:
"""Poll background I/O jobs and surface completion feedback."""
jobs = list((io_jobs or {}).get("jobs", []))
if not jobs:
return no_update, no_update, no_update, no_update
updated_jobs: list[dict[str, Any]] = []
save_feedback = no_update
export_feedback = no_update
unsaved_update = no_update
for job in jobs:
job_id = job.get("id")
if not job_id:
continue
status = get_io_job_status(job_id)
if status is None or status.get("status") == "pending":
updated_jobs.append(job)
continue
forget_io_job(job_id)
job_type = job.get("type")
if status.get("status") == "error":
error_message = status.get("error", "Unknown error")
feedback = dbc.Alert(
f"I/O error: {error_message}",
color="danger",
className="mb-0 py-2",
)
if job_type == "save":
save_feedback = _room_feedback_payload(feedback)
unsaved_update = True
elif job_type == "export":
export_feedback = _room_feedback_payload(feedback)
continue
if job_type == "save":
feedback = dbc.Alert(
f"Saved: {job.get('display_name')}",
color="success",
className="mb-0 py-2",
duration=3000,
)
save_feedback = _room_feedback_payload(feedback)
unsaved_update = False
elif job_type == "export":
feedback = dbc.Alert(
f"Exported: {job.get('display_name')}.json",
color="success",
className="mb-0 py-2",
duration=3000,
)
export_feedback = _room_feedback_payload(feedback)
if updated_jobs == jobs and save_feedback is no_update and export_feedback is no_update:
return no_update, no_update, no_update, no_update
return {"jobs": updated_jobs}, save_feedback, export_feedback, unsaved_update