115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.db import session_scope
|
|
from l4d2web.models import Blueprint, BlueprintOverlay, Overlay, Server
|
|
from l4d2web.services import host_commands
|
|
from l4d2web.services.spec_yaml import write_temp_spec
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ServerStatus:
|
|
state: str
|
|
raw_active_state: str
|
|
raw_sub_state: str
|
|
|
|
|
|
def build_server_spec_payload(server: Server, blueprint: Blueprint, overlay_refs: list[str]) -> dict:
|
|
return {
|
|
"port": server.port,
|
|
"overlays": overlay_refs,
|
|
"arguments": json.loads(blueprint.arguments),
|
|
"config": json.loads(blueprint.config),
|
|
}
|
|
|
|
|
|
def load_server_blueprint_bundle(server_id: int) -> tuple[Server, Blueprint, list[str]]:
|
|
with session_scope() as db:
|
|
server = db.scalar(select(Server).where(Server.id == server_id))
|
|
if server is None:
|
|
raise ValueError("server not found")
|
|
|
|
blueprint = db.scalar(select(Blueprint).where(Blueprint.id == server.blueprint_id))
|
|
if blueprint is None:
|
|
raise ValueError("blueprint not found")
|
|
|
|
rows = db.execute(
|
|
select(Overlay.path)
|
|
.join(BlueprintOverlay, BlueprintOverlay.overlay_id == Overlay.id)
|
|
.where(BlueprintOverlay.blueprint_id == blueprint.id)
|
|
.order_by(BlueprintOverlay.position)
|
|
).all()
|
|
overlay_refs = [row[0] for row in rows]
|
|
return server, blueprint, overlay_refs
|
|
|
|
|
|
def install_runtime(on_stdout=None, on_stderr=None, should_cancel=None) -> None:
|
|
host_commands.run_command(
|
|
["l4d2ctl", "install"],
|
|
on_stdout=on_stdout,
|
|
on_stderr=on_stderr,
|
|
should_cancel=should_cancel,
|
|
)
|
|
|
|
|
|
def initialize_server(server_id: int, on_stdout=None, on_stderr=None, should_cancel=None) -> None:
|
|
server, blueprint, overlay_refs = load_server_blueprint_bundle(server_id)
|
|
spec_path = write_temp_spec(build_server_spec_payload(server, blueprint, overlay_refs))
|
|
try:
|
|
host_commands.run_command(
|
|
["l4d2ctl", "initialize", server.name, "-f", str(spec_path)],
|
|
on_stdout=on_stdout,
|
|
on_stderr=on_stderr,
|
|
should_cancel=should_cancel,
|
|
)
|
|
finally:
|
|
spec_path.unlink(missing_ok=True)
|
|
|
|
|
|
def start_server(server_id: int, on_stdout=None, on_stderr=None, should_cancel=None) -> None:
|
|
server, _, _ = load_server_blueprint_bundle(server_id)
|
|
host_commands.run_command(
|
|
["l4d2ctl", "start", server.name],
|
|
on_stdout=on_stdout,
|
|
on_stderr=on_stderr,
|
|
should_cancel=should_cancel,
|
|
)
|
|
|
|
|
|
def stop_server(server_id: int, on_stdout=None, on_stderr=None, should_cancel=None) -> None:
|
|
server, _, _ = load_server_blueprint_bundle(server_id)
|
|
host_commands.run_command(
|
|
["l4d2ctl", "stop", server.name],
|
|
on_stdout=on_stdout,
|
|
on_stderr=on_stderr,
|
|
should_cancel=should_cancel,
|
|
)
|
|
|
|
|
|
def delete_server(server_id: int, on_stdout=None, on_stderr=None, should_cancel=None) -> None:
|
|
server, _, _ = load_server_blueprint_bundle(server_id)
|
|
host_commands.run_command(
|
|
["l4d2ctl", "delete", server.name],
|
|
on_stdout=on_stdout,
|
|
on_stderr=on_stderr,
|
|
should_cancel=should_cancel,
|
|
)
|
|
|
|
|
|
def server_status(server_name: str) -> ServerStatus:
|
|
result = host_commands.run_command(["l4d2ctl", "status", server_name, "--json"])
|
|
payload = json.loads(result.stdout or "{}")
|
|
return ServerStatus(
|
|
state=str(payload.get("state", "unknown")),
|
|
raw_active_state=str(payload.get("raw_active_state", "unknown")),
|
|
raw_sub_state=str(payload.get("raw_sub_state", "unknown")),
|
|
)
|
|
|
|
|
|
def stream_server_logs(server_name: str, *, lines: int = 200, follow: bool = True):
|
|
command = ["l4d2ctl", "logs", server_name, "--lines", str(lines)]
|
|
command.append("--follow" if follow else "--no-follow")
|
|
return host_commands.stream_command(command)
|