shawn/tactical-shooter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 17 Wiki Activity

Save workspace artifacts: character-controller, gdextension scaffold, network scripts, map-pipeline, server config

Browse Source 
Includes code from task workspaces that was never pushed:
- client/characters/ — fps_character_controller.gd (400 lines), fps_camera.gd, input_handler.gd
- gdextension/simulation/ — C++ GDExtension scaffold: entity, movement, hit detection, simulation server, state serializer, bitstream (14 files)
- scripts/network/ — ENet-based network_manager.gd, server/client_main.gd, player.gd
- scripts/map_packaging/ — PCK pipeline: pack_map.gd, map_downloader.gd, map_registry_server.py, README
- scripts/config/ — server_config.gd (480 lines)
- scenes/ — client_main, server_main, player, test_range
- project.godot, export_presets.cfg
This commit is contained in:
Shawn
2026-07-01 18:30:44 -04:00
parent 2cf57a989f
commit e9dc05983c
42 changed files with 6249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/map_packaging/map_registry_server.py
+393
View File
@@ -0,0 +1,393 @@
#!/usr/bin/env python3
"""
map_registry_server.py — Map Registry HTTP Server
Serves packaged .pck map files and a JSON map listing for the Godot
client's MapDownloader to consume.
Endpoints
---------
GET /maps
Returns a JSON list of available maps with metadata.
Response:
{
"maps": [
{
"name": "de_dust2",
"size": 4194304,
"version": 1,
"description": "Classic bomb-defusal map",
"scene": "res://scenes/maps/de_dust2.tscn",
"checksum_sha256": "a1b2c3..."
}
],
"server_name": "Tactical Shooter Map Registry",
"map_count": 1
}
GET /maps/<name>.pck
Download a packaged map file. Content-Type: application/octet-stream.
GET /maps/<name>.json
Download metadata for a specific map.
Response:
{
"name": "de_dust2",
"size": 4194304,
"version": 1,
"description": "Classic bomb-defusal map",
"scene": "res://scenes/maps/de_dust2.tscn",
"checksum_sha256": "a1b2c3...",
"packed_at": "2026-06-30 12:00:00"
}
Usage
-----
# Start the server on the default port (8090):
python3 map_registry_server.py
# Custom port and map directory:
python3 map_registry_server.py --port 8080 --maps-dir /data/maps
# Docker / container friendly:
MAP_REGISTRY_PORT=8080 MAP_REGISTRY_MAPS=/data/maps python3 map_registry_server.py
The server scans `maps_dir` (default: ./packed_maps/) for *.pck files and
their matching *.json metadata files on startup and on SIGHUP.
Integration
-----------
In your server's ServerConfig or config, add a registry section:
[map_registry]
enabled=true
url="http://maps.example.com:8090"
Or set env: MAP_REGISTRY_URL on the game server.
The game client's MapDownloader singleton will connect to this server
to fetch the map list and download .pck files.
"""
import argparse
import hashlib
import json
import logging
import os
import signal
import sys
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from typing import Dict, Optional
logger = logging.getLogger("MapRegistry")
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
DEFAULT_PORT = 8090
DEFAULT_MAPS_DIR = "packed_maps"
SERVER_NAME = "Tactical Shooter Map Registry"
VERSION = "1.0.0"
# ---------------------------------------------------------------------------
# Map Registry
# ---------------------------------------------------------------------------
class MapRegistry:
"""Scans a directory for .pck files and their metadata."""
def __init__(self, maps_dir: str):
self.maps_dir = Path(maps_dir)
self.maps_dir.mkdir(parents=True, exist_ok=True)
self._cache: Dict[str, dict] = {}
self._last_scan: float = 0
self._scan_interval: float = 5.0 # seconds between rescans
self._scan()
def _scan(self) -> None:
"""Scan the maps directory for .pck and .json files."""
now = time.time()
if now - self._last_scan < self._scan_interval:
return
self._last_scan = now
self._cache.clear()
if not self.maps_dir.exists():
logger.warning("Maps directory does not exist: %s", self.maps_dir)
return
for pck_file in sorted(self.maps_dir.glob("*.pck")):
map_name = pck_file.stem
json_file = pck_file.with_suffix(".json")
entry = {
"name": map_name,
"size": pck_file.stat().st_size,
"path": str(pck_file.relative_to(self.maps_dir)),
"version": 1,
"description": "",
"scene": f"res://scenes/maps/{map_name}.tscn",
"checksum_sha256": "",
"packed_at": "",
}
# Load metadata from sidecar JSON if it exists
if json_file.exists():
try:
with open(json_file, "r") as f:
meta = json.load(f)
entry["version"] = meta.get("version", 1)
entry["description"] = meta.get("description", "")
entry["packed_at"] = meta.get("packed_at", "")
entry["source_scene"] = meta.get("source_scene", "")
except (json.JSONDecodeError, OSError) as e:
logger.warning("Failed to parse metadata for %s: %s", map_name, e)
# Compute SHA-256 checksum (on first scan, cached in memory)
ck = self._compute_checksum(pck_file)
if ck:
entry["checksum_sha256"] = ck
self._cache[map_name] = entry
logger.debug("Registered map: %s (%d bytes)", map_name, entry["size"])
logger.info("Scanned %s: %d maps registered", self.maps_dir, len(self._cache))
def _compute_checksum(self, path: Path) -> str:
"""Compute SHA-256 checksum of a file."""
try:
h = hashlib.sha256()
with open(path, "rb") as f:
while True:
chunk = f.read(65536)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
except OSError as e:
logger.warning("Checksum failed for %s: %s", path.name, e)
return ""
def get_map_list(self) -> dict:
"""Return the full map listing as a dict."""
self._scan()
return {
"maps": list(self._cache.values()),
"server_name": SERVER_NAME,
"server_version": VERSION,
"map_count": len(self._cache),
}
def get_map_info(self, name: str) -> Optional[dict]:
"""Return metadata for a single map."""
self._scan()
return self._cache.get(name)
def get_map_file(self, name: str) -> Optional[Path]:
"""Return the filesystem path to a .pck file, or None."""
self._scan()
entry = self._cache.get(name)
if entry is None:
return None
pck_path = self.maps_dir / entry["path"]
return pck_path if pck_path.exists() else None
# ---------------------------------------------------------------------------
# HTTP Handler
# ---------------------------------------------------------------------------
class MapRegistryHandler(BaseHTTPRequestHandler):
"""HTTP request handler for the map registry API."""
# Shared across all instances (set by the server)
registry: MapRegistry = None # type: ignore
def log_message(self, format: str, *args) -> None:
logger.info("%s - %s", self.client_address[0], format % args)
def _send_json(self, data: dict, status: int = 200) -> None:
body = json.dumps(data, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(body)
def _send_error(self, status: int, message: str) -> None:
self._send_json({"error": message}, status)
def _send_file(self, path: Path, content_type: str = "application/octet-stream") -> None:
try:
file_size = path.stat().st_size
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(file_size))
self.send_header("Content-Disposition", f'attachment; filename="{path.name}"')
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
with open(path, "rb") as f:
while True:
chunk = f.read(65536)
if not chunk:
break
self.wfile.write(chunk)
except OSError as e:
logger.error("File send failed for %s: %s", path.name, e)
self._send_error(500, "Internal server error")
def do_GET(self) -> None:
path = self.path.rstrip("/")
# GET /maps — list all available maps
if path == "/maps":
self._send_json(self.registry.get_map_list())
return
# GET /maps/<name>.pck — download a map file
if path.startswith("/maps/") and path.endswith(".pck"):
map_name = path[len("/maps/"):-len(".pck")]
map_file = self.registry.get_map_file(map_name)
if map_file:
self._send_file(map_file)
else:
self._send_error(404, f"Map '{map_name}' not found")
return
# GET /maps/<name>.json — get metadata for a single map
if path.startswith("/maps/") and path.endswith(".json"):
map_name = path[len("/maps/"):-len(".json")]
info = self.registry.get_map_info(map_name)
if info:
self._send_json(info)
else:
self._send_error(404, f"Map '{map_name}' not found")
return
# GET / — server info
if path == "/" or path == "":
self._send_json({
"service": SERVER_NAME,
"version": VERSION,
"endpoints": {
"list_maps": "/maps",
"download_map": "/maps/<name>.pck",
"map_metadata": "/maps/<name>.json",
},
})
return
self._send_error(404, f"Not found: {path}")
def do_OPTIONS(self) -> None:
"""CORS preflight."""
self.send_response(204)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
# ---------------------------------------------------------------------------
# Server
# ---------------------------------------------------------------------------
def create_server(port: int, maps_dir: str) -> HTTPServer:
"""Create and configure the HTTP server."""
registry = MapRegistry(maps_dir)
MapRegistryHandler.registry = registry
server = HTTPServer(("0.0.0.0", port), MapRegistryHandler)
server.timeout = 0.5 # allow signal handling
logger.info("Map Registry Server v%s", VERSION)
logger.info(" Listen: http://0.0.0.0:%d", port)
logger.info(" Maps dir: %s", registry.maps_dir.resolve())
logger.info(" Maps found: %d", len(registry._cache))
logger.info(" Endpoints:")
logger.info(" List: GET /maps")
logger.info(" Download: GET /maps/<name>.pck")
logger.info(" Metadata: GET /maps/<name>.json")
return server
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Tactical Shooter Map Registry Server",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
parser.add_argument(
"--port", "-p",
type=int,
default=int(os.environ.get("MAP_REGISTRY_PORT", DEFAULT_PORT)),
help="HTTP port (default: %d)" % DEFAULT_PORT,
)
parser.add_argument(
"--maps-dir", "-d",
type=str,
default=os.environ.get("MAP_REGISTRY_MAPS", DEFAULT_MAPS_DIR),
help="Directory containing .pck files (default: %s)" % DEFAULT_MAPS_DIR,
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="Enable debug logging",
)
return parser.parse_args()
def main():
args = parse_args()
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
server = create_server(args.port, args.maps_dir)
# Handle SIGTERM/SIGINT for graceful shutdown
shutdown_requested = False
def handle_signal(sig, frame):
nonlocal shutdown_requested
if not shutdown_requested:
logger.info("Shutdown requested (signal %d)", sig)
shutdown_requested = True
server.shutdown()
signal.signal(signal.SIGTERM, handle_signal)
signal.signal(signal.SIGINT, handle_signal)
# SIGHUP rescans the map directory
if hasattr(signal, "SIGHUP"):
def handle_hup(sig, frame):
logger.info("SIGHUP received — rescanning maps")
server.RequestHandlerClass.registry._scan()
signal.signal(signal.SIGHUP, handle_hup)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
logger.info("Server stopped")
if __name__ == "__main__":
main()