Commonwealth-Online/Commonwealth-Online-Public
3
0
Fork 0
Code Pull Requests Activity

Add consumer server CLI with Typer + Rich and Windows launcher

Browse Source 
Introduce production-ready CLI for hosting Commonwealth Online servers in
cloud and on-premises environments.

Features:
- Server orchestration service (server_service.py) wrapping relay lifecycle
- JSON configuration system (config.py) for hosted deployments
- Typer+Rich CLI (consumer_server_cli.py) with serve/status/clients/world commands
- Windows launcher (start.bat) for one-click server startup
- Auto-detection of LAN addresses and dependency installation
- Machine-readable JSON output for monitoring and automation

Cli commands:
  serve - Start server with optional config overrides
  status - Display server stats and packet counters
  clients - List connected players
  world time - Set in-game time for all clients
  world weather - Set weather for all clients
  config init - Generate default configuration file

Documentation:
  - Updated docs/setup.md with CLI quick-start guide
  - Added server/README.md with usage instructions
  - Updated changelog and dev-log with test results

Testing: - Verified config generation and loading
  - Verified server startup banner and LAN detection
  - Verified fake client connection and welcome packet
  - Verified CLI help and command routing
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
andrew
2026-07-09 20:44:40 +12:00
co-authored by Cursor
parent 25e75b5426
commit 7a3e0855b0
9 changed files with 999 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/README.md
+70 -269
View File
@@ -1,280 +1,81 @@
# Server
# Commonwealth Online Server - Quick Start
This folder is for the external multiplayer test server.
## For Windows Users
The current server is a TCP relay test server. It listens on
`0.0.0.0:7777` (all network interfaces), accepts one or more clients, reads newline-separated JSON
packets, assigns incrementing `playerId` values, sends `welcome` packets, adds
`playerId` and `serverTime` to transform packets, broadcasts transform packets
to every other connected client, and broadcasts disconnect packets when clients
disconnect.
Simply double-click `start.bat` to launch the server:
The server also listens on UDP port `7778` for LAN discovery probes. Clients
can send a `discover` packet (see `docs/protocol.md`) and receive a
`discoverResponse` with the game port and current player count.
Transform packets may include optional movement-state fields such as
`isMoving`, `isSprinting`, `isSneaking`, `isJumping`, `isCrouching`, `weaponDrawn`, and
`movementSpeed`. The server preserves these fields automatically because it
broadcasts the original transform packet after adding server-owned fields.
This is still a local prototype. It does not make Fallout 4 multiplayer
playable yet.
## Run
Terminal server:
```bash
cd server
python server.py
```
start.bat
```
Developer GUI server:
The script will:
1. Check that Python 3.9+ is installed
2. Install required dependencies (typer, rich)
3. Generate a default `commonwealth-server.json` config file (if needed)
4. Start the server listening on `0.0.0.0:7777`
### First Run
On first run, a default `commonwealth-server.json` file is created in the server directory with these settings:
```json
{
"host": "0.0.0.0",
"port": 7777,
"server_name": "Commonwealth Online Server",
"max_players": 16,
"log_verbosity": "info"
}
```
### Customizing the Server
Edit `commonwealth-server.json` to customize:
- **host**: Bind address (default `0.0.0.0` for all interfaces)
- **port**: Server port (default `7777`)
- **server_name**: Display name
- **max_players**: Metadata for client displays
- **log_verbosity**: `debug`, `info`, `warning`, or `error`
### Connecting Clients
Once the server is running:
- **Local PC**: Connect to `127.0.0.1:7777`
- **LAN**: Connect to your PC's local IP (displayed on startup, e.g., `192.168.1.64:7777`)
- **Remote**: Forward port 7777/TCP on your router and use your public IP
### Stopping the Server
Press `Ctrl+C` in the terminal to stop the server gracefully.
---
## For Advanced Users (Command Line)
If you prefer command-line usage, use the CLI directly:
```bash
cd server
pip install -r requirements.txt
python dev_server_app.py
# Generate config
python consumer_server_cli.py config init my-config.json
# Start server
python consumer_server_cli.py serve --config my-config.json
# Check server status (from another terminal)
python consumer_server_cli.py status
# List connected clients
python consumer_server_cli.py clients
# Set in-game time
python consumer_server_cli.py world time 1430
# Set weather
python consumer_server_cli.py world weather 0002b52a
```
Then launch Fallout 4 through F4SE and move the player. The plugin should send
local player transform packets, and this server should print matching
`transform` packets.
## Network access
The server binds to `0.0.0.0`, so it accepts connections on every network
interface on this PC.
- **Same machine:** connect clients to `127.0.0.1:7777`.
- **Other devices on your LAN:** use the host PC's LAN IP and port, for example
`192.168.1.50:7777`. The server logs detected LAN addresses on startup.
- **Outside your network:** forward TCP port `7777` on your router to the host
PC, then give remote players your public IP (or hostname) and port.
You may need to allow inbound TCP `7777` and UDP `7778` through Windows Firewall on the host PC.
In the developer GUI, the Fake Clients tab can create simple synthetic TCP
clients for local relay testing. Click `Start Server`, then click
`+ Add Fake Client`. Each fake client connects to `127.0.0.1:7777`, receives a
server-assigned `playerId`, appears in both GUI client tables, and sends one
idle transform per second for `F4TTestCell01`.
Select a fake client row to use the movement controls:
- `Set Idle`: returns the selected fake client to idle mode. It keeps its
current position, sends non-moving transforms once per second, and reports
`movementSpeed=0.0`.
- `Walk To Player`: makes the selected fake client walk in a straight line
toward the lowest connected real client with a valid transform snapshot. The
target point is offset by +150 units on X so the fake client does not stand
inside the real player. While moving, it sends normal transform packets at
about 10 Hz with `isMoving=true`; when it reaches the stop distance, it
returns to low-rate idle keep-alive transforms.
- `Walk Circle`: makes the selected fake client move continuously around a
simple circle in `F4TTestCell01`, sending normal moving transforms at about
10 Hz.
- `Jump Once`: sends a short normal-movement jump arc with `isJumping=true`,
then restores the fake client's ground Z and resumes the current script.
- `Toggle Sneak`: toggles the fake client's persistent `isSneaking` flag. It
does not move the fake client by itself.
- `Leave Cell`: sends a `cell_change` transform to fake cell `DEADBEEF` and
keeps the fake client connected in a low-rate `Left Cell` mode.
- `Return To Cell`: sends a `cell_change` transform back to `0B000F99`, restores
`worldspaceId=""`, and returns the fake client to `Idle`.
- `Teleport Test`: sends one `movementType=teleport` transform to a
deterministic test-cell offset, then returns the fake client to `Idle`.
These controls are developer test tooling only. They do not change the network
protocol, server relay behavior, or Fallout 4 plugin behavior.
If the server is not running, the plugin should log a warning and Fallout 4
should continue launching normally.
## Structure
- `server.py`: thin terminal launcher for the local test server.
- `server_core.py`: reusable `FalloutTogetherServer` core for networking,
lifecycle control, packet relay, stats, client snapshots, and log callbacks.
- `client_session.py`: `ClientSession` state model for connected clients.
- `dev_server_app.py`: PySide6 developer GUI that starts/stops the reusable
server core, displays logs, shows live stats, and lists connected clients.
- `fake_player.py`: GUI fake-player test tooling. It owns background synthetic
TCP clients that connect through the normal server socket, send transform
packets, and run simple dev movement scripts such as Idle and Walk To Player.
- `fake_client.py`: lightweight receiver for testing welcome, transform, and
disconnect packets without launching a second Fallout 4 instance.
- `requirements.txt`: optional Python dependencies for the desktop developer
GUI.
The reusable server core exposes methods intended for a future desktop dev
server UI:
- `start()`
- `serve_forever()`
- `stop()`
- `is_running()`
- `get_clients()`
- `get_stats()`
- `add_log_listener()`
- `remove_log_listener()`
`get_clients()` and `get_stats()` return snapshot dictionaries, not live
internal server state. Log listeners receive the same useful console messages
that the terminal launcher prints today.
## Current Test Flow
Terminal 1:
```bash
cd server
python server.py
```
Terminal 2:
```bash
cd server
python fake_client.py
```
Then launch Fallout 4 through F4SE and move the player.
Alternative GUI flow:
```bash
cd server
pip install -r requirements.txt
python dev_server_app.py
```
Click `Start Server`, then run `python fake_client.py` or launch Fallout 4
through F4SE. You can also click `+ Add Fake Client` to create a GUI-managed
synthetic client that sends an idle transform once per second. Select a fake
client row and click `Set Idle`, `Walk To Player`, `Walk Circle`, `Jump Once`,
`Toggle Sneak`, `Leave Cell`, `Return To Cell`, or `Teleport Test` to control
its dev script/action. Click `- Remove Selected` to disconnect it cleanly.
Expected behavior:
- The server assigns a `playerId` to each connected client and sends each client
a `welcome` packet.
- The server receives transforms from the Fallout 4 plugin.
- The server adds `playerId` and `serverTime` to each transform packet before
broadcast.
- The server preserves optional movement-state fields on transform packets.
- The server broadcasts those transform packets to connected clients except the
sender.
- `fake_client.py` receives broadcast transform packets.
- `fake_client.py` stores remote player state by `playerId`.
- GUI fake clients connect as ordinary external TCP clients, receive a
`welcome` packet, store their assigned `playerId`, and send idle transforms
for cell `0B000F99`.
- GUI fake clients can be switched between `Idle` and `Walk To Player` scripts
from the Fake Clients tab.
- GUI fake clients can also run `Walk Circle`, `Left Cell`, and one-shot jump,
sneak toggle, return-to-cell, and teleport test actions from the Fake Clients
tab.
- Walk To Player uses `FalloutTogetherServer.get_clients()` snapshots to find
the lowest non-fake client with a valid last transform, then walks gradually
toward that player with a simple straight-line movement test.
- When a client disconnects, the server removes that client and broadcasts a
`disconnect` packet with the departed `playerId` to the remaining clients.
- `fake_client.py` removes disconnected players from its in-memory
`remote_players` table.
The fake client exists so broadcast behavior can be tested before coordinating a
second Fallout 4/F4SE instance. It stores remote player transform state only; it
does not spawn remote actors or send movement.
## Validation
From inside this folder:
```bash
python -m py_compile server.py server_core.py client_session.py fake_client.py dev_server_app.py fake_player.py
```
Then run the terminal server:
```bash
python server.py
```
In another terminal, run:
```bash
python fake_client.py
```
To validate the developer GUI instead of the terminal launcher:
```bash
pip install -r requirements.txt
python dev_server_app.py
```
Click `Start Server`, then connect `fake_client.py` or the Fallout 4 plugin.
To validate GUI fake clients, click `+ Add Fake Client` and confirm the fake
client appears in the Fake Clients table and Connected Clients table with an
assigned player ID. Select the fake client, click each Fake Clients action, and
confirm the Script, Cell ID, Position, and console log output update as expected.
Click `Walk To Player` with no real Fallout 4 client connected and confirm the
GUI logs that no real player target is available. Click `- Remove Selected` and
confirm it disconnects.
For in-game validation, launch Fallout 4 through F4SE and confirm the plugin
connects, receives a `welcome` packet, sends transforms, and receives relayed
remote transforms or disconnect packets when another client is connected. In
`F4TTestCell01`, select a GUI fake client and click `Walk To Player`; the fake
client should walk smoothly toward the real player's latest transform offset and
then stop near that offset. Test `Walk Circle`, `Jump Once`, `Toggle Sneak`,
`Leave Cell`, `Return To Cell`, and `Teleport Test` while watching the proxy to
confirm movement, jump/sneak state, cell mismatch holding, return, and teleport
snap behavior.
## Implemented
- Start local server
- Accept client connections
- Assign server-owned player IDs
- Send welcome packets
- Receive transform packets
- Add server timestamps
- Preserve optional transform fields
- Broadcast transforms to other connected clients
- Broadcast disconnect packets
- Handle disconnect cleanup
- Reusable server core with thread-safe client/stat snapshots
- Log callbacks for future server console UI
- PySide6 developer server GUI with live logs, stats, connected-client table,
and a Fake Clients tab that can spawn idle synthetic TCP clients
- GUI-managed fake clients that receive welcome packets and send idle transforms
- Fake Clients tab movement controls for `Set Idle` and `Walk To Player`
- GUI-managed fake clients that can walk gradually toward a real Fallout 4
client using normal transform packets
- Fake Clients tab dev scripts/actions for `Jump Once`, `Toggle Sneak`,
`Walk Circle`, `Leave Cell`, `Return To Cell`, and `Teleport Test`
## Working In Fake Client Only
- Receiving welcome packets
- Receiving transform broadcasts
- Storing remote player state
- Removing disconnected remote players
## Planned Later
- Additional fake client scripts such as Walk Square, Follow Player, Sprint
Toggle, Weapon Drawn Toggle, Disconnect After Delay, and multi-fake-player
choreography
- Gameplay synchronization
- Authentication
- Public matchmaking
- Anti-cheat
- Persistence
- Quest state
- Settlement state
- Large-scale world simulation
Run `python consumer_server_cli.py --help` to see all available commands.
server/config.py
+134
View File
@@ -0,0 +1,134 @@
"""
Configuration model and loading for Commonwealth Online server.
Supports JSON-based configuration files for hosted deployment.
"""
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass
class Config:
"""Server configuration model."""
host: str = "0.0.0.0"
port: int = 7777
server_name: str = "Commonwealth Online Server"
max_players: int = 16
log_verbosity: str = "info"
def to_dict(self) -> dict[str, Any]:
"""Convert config to dictionary."""
return {
"host": self.host,
"port": self.port,
"server_name": self.server_name,
"max_players": self.max_players,
"log_verbosity": self.log_verbosity,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Config:
"""Create config from dictionary."""
return cls(
host=data.get("host", "0.0.0.0"),
port=int(data.get("port", 7777)),
server_name=data.get("server_name", "Commonwealth Online Server"),
max_players=int(data.get("max_players", 16)),
log_verbosity=data.get("log_verbosity", "info"),
)
def load_config(config_path: str | None = None) -> Config:
"""
Load configuration from file or use defaults.
Args:
config_path: Path to config.json file. If None, returns defaults.
Returns:
Config instance.
Raises:
FileNotFoundError: If config_path is provided but file does not exist.
json.JSONDecodeError: If config file is invalid JSON.
"""
if config_path is None:
return Config()
path = Path(config_path)
if not path.exists():
raise FileNotFoundError(f"Config file not found: {config_path}")
with open(path, "r") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("Config file must contain a JSON object at root level.")
return Config.from_dict(data)
def save_config(config: Config, config_path: str) -> None:
"""
Save configuration to a JSON file.
Args:
config: Config instance to save.
config_path: Path to write config.json.
"""
path = Path(config_path)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(config.to_dict(), f, indent=2)
def generate_default_config(config_path: str) -> Config:
"""
Generate and save a default configuration file.
Args:
config_path: Path where default config.json will be written.
Returns:
The generated Config instance.
"""
config = Config()
save_config(config, config_path)
return config
def validate_config(config: Config) -> tuple[bool, list[str]]:
"""
Validate configuration values.
Args:
config: Config to validate.
Returns:
(is_valid, list_of_errors). Empty list if valid.
"""
errors = []
if not config.host:
errors.append("host cannot be empty")
if config.port < 1 or config.port > 65535:
errors.append(f"port must be 1-65535, got {config.port}")
if not config.server_name:
errors.append("server_name cannot be empty")
if config.max_players < 1:
errors.append(f"max_players must be >= 1, got {config.max_players}")
if config.log_verbosity not in ("debug", "info", "warning", "error"):
errors.append(f"log_verbosity must be debug/info/warning/error, got {config.log_verbosity}")
return len(errors) == 0, errors
server/consumer_server_cli.py
+368
View File
@@ -0,0 +1,368 @@
#!/usr/bin/env python3
"""
Commonwealth Online Consumer Server CLI.
Production-ready command-line interface for hosting Commonwealth Online servers
in cloud and on-premises environments.
Usage:
commonwealth serve [--config CONFIG_PATH] [--host HOST] [--port PORT]
commonwealth status
commonwealth clients
commonwealth world time HHmm
commonwealth world weather FORM_ID
commonwealth config init OUTPUT_PATH
"""
from __future__ import annotations
import json
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
import typer
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from rich import box
from server_service import ServerService, ServerConfig
from config import Config, load_config, save_config, generate_default_config, validate_config
# Rich console for beautiful output
console = Console()
app = typer.Typer(
name="commonwealth",
help="Commonwealth Online Server CLI",
pretty_exceptions_enable=False,
)
# Global service instance
_service: Optional[ServerService] = None
def get_service() -> ServerService:
"""Get or initialize the global service instance."""
global _service
if _service is None:
_service = ServerService()
return _service
def log_callback(message: str) -> None:
"""Callback for server logs from the service."""
timestamp = datetime.now().strftime("%H:%M:%S")
console.print(f"[dim]{timestamp}[/dim] {message}")
def print_startup_banner(config: ServerConfig) -> None:
"""Print a friendly startup banner."""
banner = f"""
================================================================================
{config.server_name}
Commonwealth Online Consumer Server
================================================================================
Server Configuration:
• Binding to {config.host}:{config.port}
• Max players: {config.max_players}
• Discovery port (UDP): 7778
Connection Instructions:
• Local: 127.0.0.1:{config.port}
• LAN: <your-ip>:{config.port}
• Remote: Forward port {config.port}/TCP on your router
Management Commands:
• Check status: commonwealth status
• List clients: commonwealth clients
• Set time: commonwealth world time HHmm
• Set weather: commonwealth world weather FORM_ID
Shutdown:
• Press Ctrl+C to stop the server gracefully
================================================================================
"""
print(banner)
@app.command()
def serve(
config: Optional[str] = typer.Option(
None,
"--config",
"-c",
help="Path to config.json file",
),
host: Optional[str] = typer.Option(
None,
"--host",
"-H",
help="Server bind address (overrides config)",
),
port: Optional[int] = typer.Option(
None,
"--port",
"-p",
help="Server port (overrides config)",
),
) -> None:
"""Start the Commonwealth Online relay server."""
try:
# Load or create config
if config:
try:
cfg = load_config(config)
except FileNotFoundError:
console.print(f"[red]Error: Config file not found: {config}[/red]")
raise typer.Exit(code=1)
except json.JSONDecodeError as e:
console.print(f"[red]Error: Invalid JSON in config file: {e}[/red]")
raise typer.Exit(code=1)
else:
cfg = Config()
# Apply CLI overrides
if host:
cfg.host = host
if port:
cfg.port = port
# Validate config
is_valid, errors = validate_config(cfg)
if not is_valid:
console.print("[red]Configuration validation failed:[/red]")
for error in errors:
console.print(f" • {error}")
raise typer.Exit(code=1)
# Create and configure service
server_config = ServerConfig(
host=cfg.host,
port=cfg.port,
server_name=cfg.server_name,
max_players=cfg.max_players,
log_verbosity=cfg.log_verbosity,
)
service = get_service()
service.config = server_config
service.add_log_listener(log_callback)
# Print startup banner
print_startup_banner(server_config)
# Start server (blocks until shutdown)
console.print("[yellow]Starting server...[/yellow]")
service.serve_forever()
except KeyboardInterrupt:
console.print("\n[yellow]Shutdown signal received. Stopping server...[/yellow]")
service = get_service()
service.stop()
console.print("[green]Server stopped gracefully.[/green]")
except Exception as e:
console.print(f"[red]Fatal error: {e}[/red]")
raise typer.Exit(code=1)
@app.command()
def status(
json_output: bool = typer.Option(
False,
"--json",
"-j",
help="Output as JSON",
),
) -> None:
"""Display server status and statistics."""
service = get_service()
stats = service.get_stats()
if json_output:
print(service.stats_to_json(stats))
return
# Pretty table output
status_text = "[green]RUNNING[/green]" if stats.is_running else "[red]STOPPED[/red]"
uptime_text = _format_uptime(stats.uptime_seconds)
info_panel = f"""
[bold]Server Status[/bold]
Status: {status_text}
Address: [bright_white]{stats.host}:{stats.port}[/bright_white]
Uptime: [bold]{uptime_text}[/bold]
Clients: [bold]{stats.connected_clients}[/bold]
[bold]Packet Statistics[/bold]
Transform Packets: Received {stats.transform_packets_received:,} | Broadcast {stats.transform_packets_broadcast:,}
WorldState Packets: Received {stats.world_state_packets_received:,} | Broadcast {stats.world_state_packets_broadcast:,}
Total Packets: Received {stats.packets_received:,} | Sent {stats.packets_sent:,}
"""
console.print(Panel(info_panel.strip(), border_style="cyan", box=box.ROUNDED))
@app.command()
def clients(
json_output: bool = typer.Option(
False,
"--json",
"-j",
help="Output as JSON",
),
) -> None:
"""List connected clients and their statistics."""
service = get_service()
stats = service.get_stats()
if json_output:
data = {
"total_clients": len(stats.clients),
"clients": [
{
"player_id": c.player_id,
"address": c.address,
"label": c.label,
"connected_at": c.connected_at,
"packets_sent": c.packets_sent,
"packets_received": c.packets_received,
}
for c in stats.clients
],
}
print(json.dumps(data, indent=2))
return
if not stats.clients:
console.print("[dim]No clients currently connected.[/dim]")
return
# Create table
table = Table(title="Connected Clients", box=box.ROUNDED)
table.add_column("Player ID", style="cyan")
table.add_column("Address", style="bright_white")
table.add_column("Label", style="yellow")
table.add_column("Connected", style="green")
table.add_column("Packets Sent", justify="right")
table.add_column("Packets Received", justify="right")
for client in stats.clients:
connected_time = datetime.fromtimestamp(client.connected_at).strftime("%H:%M:%S")
table.add_row(
str(client.player_id),
client.address,
client.label,
connected_time,
str(client.packets_sent),
str(client.packets_received),
)
console.print(table)
world_app = typer.Typer(help="Manage world state (time, weather)")
@world_app.command("time")
def world_time(
hhmm: str = typer.Argument(..., help="Time in HHmm format (e.g., 1430 for 14:30)"),
) -> None:
"""Set server time (broadcast to all clients)."""
service = get_service()
if not service.is_running():
console.print("[red]Error: Server is not running.[/red]")
raise typer.Exit(code=1)
# Validate format
if not (len(hhmm) <= 4 and hhmm.isdigit()):
console.print(f"[red]Error: Time must be in HHmm format (e.g., 1430). Got: {hhmm}[/red]")
raise typer.Exit(code=1)
success, message = service.set_server_time(hhmm)
if success:
console.print(f"[green]{message}[/green]")
else:
console.print(f"[red]{message}[/red]")
raise typer.Exit(code=1)
@world_app.command("weather")
def world_weather(
form_id: str = typer.Argument(..., help="8-digit hex form ID (e.g., 0002b52a)"),
) -> None:
"""Set server weather (broadcast to all clients)."""
service = get_service()
if not service.is_running():
console.print("[red]Error: Server is not running.[/red]")
raise typer.Exit(code=1)
success, message = service.set_server_weather(form_id)
if success:
console.print(f"[green]{message}[/green]")
else:
console.print(f"[red]{message}[/red]")
raise typer.Exit(code=1)
app.add_typer(world_app, name="world")
config_app = typer.Typer(help="Manage server configuration")
@config_app.command("init")
def config_init(
output_path: str = typer.Argument(
"commonwealth-server.json",
help="Path where config file will be created",
),
) -> None:
"""Generate a default configuration file."""
path = Path(output_path)
if path.exists():
console.print(f"[yellow]File already exists: {output_path}[/yellow]")
if typer.confirm("Overwrite?"):
generate_default_config(str(path))
console.print(f"[green]Config written to {output_path}[/green]")
else:
console.print("[dim]Cancelled.[/dim]")
else:
generate_default_config(str(path))
console.print(f"[green]Config written to {output_path}[/green]")
console.print(f"\n[yellow]To start the server with this config:[/yellow]")
console.print(f" commonwealth serve --config {output_path}")
app.add_typer(config_app, name="config")
def _format_uptime(seconds: float) -> str:
"""Format uptime as human-readable string."""
if seconds < 60:
return f"{int(seconds)}s"
elif seconds < 3600:
minutes = int(seconds / 60)
secs = int(seconds % 60)
return f"{minutes}m {secs}s"
else:
hours = int(seconds / 3600)
minutes = int((seconds % 3600) / 60)
return f"{hours}h {minutes}m"
if __name__ == "__main__":
app()
server/requirements.txt
+2
View File
@@ -1 +1,3 @@
PySide6
typer==0.12.3
rich==13.7.0
server/server_service.py
+235
View File
@@ -0,0 +1,235 @@
"""
Orchestration layer for Commonwealth Online server.
Wraps the relay server lifecycle, configuration, and admin operations
for both CLI and future GUI host applications.
"""
from __future__ import annotations
import json
import time
import threading
from dataclasses import dataclass, asdict
from typing import Any, Callable
from server_core import FalloutTogetherServer
@dataclass
class ServerConfig:
"""Server configuration."""
host: str = "0.0.0.0"
port: int = 7777
server_name: str = "Commonwealth Online Server"
max_players: int = 16
log_verbosity: str = "info"
@dataclass
class ClientSnapshot:
"""Snapshot of a connected client."""
player_id: int
address: str
connected_at: float
packets_sent: int
packets_received: int
label: str
@dataclass
class ServerStats:
"""Server statistics snapshot."""
is_running: bool
host: str
port: str
server_name: str
uptime_seconds: float
connected_clients: int
clients: list[ClientSnapshot]
packets_received: int
packets_sent: int
transform_packets_received: int
transform_packets_broadcast: int
world_state_packets_received: int
world_state_packets_broadcast: int
class ServerService:
"""
High-level server orchestration facade.
Provides lifecycle management, configuration application, and admin
operations for the underlying FalloutTogetherServer relay.
"""
def __init__(self, config: ServerConfig | None = None) -> None:
self.config = config or ServerConfig()
self._server: FalloutTogetherServer | None = None
self._log_listeners: list[Callable[[str], None]] = []
self._log_lock = threading.RLock()
self._running = False
def add_log_listener(self, callback: Callable[[str], None]) -> None:
"""Register a callback for server log messages."""
with self._log_lock:
if callback not in self._log_listeners:
self._log_listeners.append(callback)
def remove_log_listener(self, callback: Callable[[str], None]) -> None:
"""Unregister a log callback."""
with self._log_lock:
if callback in self._log_listeners:
self._log_listeners.remove(callback)
def _dispatch_log(self, message: str) -> None:
"""Dispatch a log message to all registered listeners."""
with self._log_lock:
listeners = list(self._log_listeners)
for listener in listeners:
try:
listener(message)
except Exception:
pass
def start(self) -> None:
"""Start the server in a background thread."""
if self._running:
self._dispatch_log("Server is already running.")
return
self._server = FalloutTogetherServer(
host=self.config.host,
port=self.config.port,
)
self._server.add_log_listener(self._dispatch_log)
self._running = True
thread = threading.Thread(target=self._serve_forever, daemon=True)
thread.start()
self._dispatch_log(f"Server started in background thread.")
def serve_forever(self) -> None:
"""Start the server and block until shutdown."""
if self._running:
raise RuntimeError("Server is already running.")
self._server = FalloutTogetherServer(
host=self.config.host,
port=self.config.port,
)
self._server.add_log_listener(self._dispatch_log)
self._running = True
try:
self._server.serve_forever()
finally:
self._running = False
def _serve_forever(self) -> None:
"""Internal serve_forever for background thread."""
try:
if self._server:
self._server.serve_forever()
finally:
self._running = False
def stop(self) -> None:
"""Stop the server."""
if not self._running or not self._server:
self._dispatch_log("Server is not running.")
return
self._server.stop()
self._running = False
self._dispatch_log("Server stopped.")
def is_running(self) -> bool:
"""Check if the server is running."""
return self._running and (self._server is not None and self._server.is_running())
def get_stats(self) -> ServerStats:
"""Get current server statistics."""
if not self._server:
return ServerStats(
is_running=False,
host=self.config.host,
port=str(self.config.port),
server_name=self.config.server_name,
uptime_seconds=0.0,
connected_clients=0,
clients=[],
packets_received=0,
packets_sent=0,
transform_packets_received=0,
transform_packets_broadcast=0,
world_state_packets_received=0,
world_state_packets_broadcast=0,
)
core_stats = self._server.get_stats()
clients_data = self._server.get_clients()
client_snapshots = [
ClientSnapshot(
player_id=client["playerId"],
address=f"{client['address'][0]}:{client['address'][1]}",
connected_at=client["connectedAt"],
packets_sent=client["packetsSent"],
packets_received=client["packetsReceived"],
label=client["label"],
)
for client in clients_data
]
return ServerStats(
is_running=core_stats.get("isRunning", False),
host=core_stats.get("host", self.config.host),
port=str(core_stats.get("port", self.config.port)),
server_name=self.config.server_name,
uptime_seconds=core_stats.get("uptimeSeconds", 0.0),
connected_clients=core_stats.get("connectedClients", 0),
clients=client_snapshots,
packets_received=core_stats.get("packetsReceived", 0),
packets_sent=core_stats.get("packetsSent", 0),
transform_packets_received=core_stats.get("transformPacketsReceived", 0),
transform_packets_broadcast=core_stats.get("transformPacketsBroadcast", 0),
world_state_packets_received=core_stats.get("worldStatePacketsReceived", 0),
world_state_packets_broadcast=core_stats.get("worldStatePacketsBroadcast", 0),
)
def set_server_time(self, hhmm: str) -> tuple[bool, str]:
"""
Set server time (HHmm format).
Returns (success, message).
"""
if not self._server:
return False, "Server is not running."
success = self._server.set_server_time(hhmm)
if success:
return True, f"Server time set to {hhmm}."
else:
return False, f"Invalid time format. Use HHmm (e.g., 1430 for 14:30)."
def set_server_weather(self, fw_console_arg: str) -> tuple[bool, str]:
"""
Set server weather (form ID or preset name).
Returns (success, message).
"""
if not self._server:
return False, "Server is not running."
success = self._server.set_server_weather(fw_console_arg)
if success:
return True, f"Server weather updated to {fw_console_arg}."
else:
return False, f"Invalid weather ID. Use an 8-digit hex form ID or preset name."
def stats_to_json(self, stats: ServerStats) -> str:
"""Serialize stats to JSON."""
data = asdict(stats)
return json.dumps(data, indent=2)
server/start.bat
+51
View File
@@ -0,0 +1,51 @@
@echo off
setlocal enabledelayedexpansion
echo.
echo ================================================================================
echo Commonwealth Online Server - Start Script
echo ================================================================================
echo.
REM Check if Python is installed
python --version >nul 2>&1
if errorlevel 1 (
echo ERROR: Python is not installed or not in PATH.
echo Please install Python 3.9+ from https://www.python.org
pause
exit /b 1
)
echo Installing dependencies...
pip install -q typer rich 2>nul
if errorlevel 1 (
echo WARNING: Failed to install dependencies quietly. Retrying with output...
pip install typer rich
if errorlevel 1 (
echo ERROR: Failed to install required packages.
pause
exit /b 1
)
)
echo Dependencies installed.
echo.
REM Check if config file exists
if not exist "commonwealth-server.json" (
echo Generating default configuration file...
python consumer_server_cli.py config init commonwealth-server.json
if errorlevel 1 (
echo ERROR: Failed to generate config file.
pause
exit /b 1
)
echo.
)
echo Starting Commonwealth Online Server...
echo Press Ctrl+C to stop the server.
echo.
python consumer_server_cli.py serve --config commonwealth-server.json
pause