Talon/chat_grid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add clock item type with timezone/time-format and emit sound

Browse Source
This commit is contained in:
Jage9
2026-02-21 16:01:40 -05:00
parent b52f9b7862
commit b2c3f75ae3
13 changed files with 218 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
server/app/item_catalog.py
View File

@@ -3,14 +3,21 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
ItemType = Literal["radio_station", "dice", "wheel"]
ItemType = Literal["radio_station", "dice", "wheel", "clock"]
CLOCK_DEFAULT_TIME_ZONE = "America/Detroit"
CLOCK_TIME_ZONE_OPTIONS: tuple[str, ...] = (
"America/Detroit",
"America/New_York",
"America/Indiana/Indianapolis",
"America/Kentucky/Louisville",
)
@dataclass(frozen=True)
class ItemDefinition:
default_title: str
capabilities: tuple[str, ...]
use_sound: str | None
emit_sound: str | None
default_params: dict
use_cooldown_ms: int = 1000
@@ -19,22 +26,28 @@ ITEM_DEFINITIONS: dict[ItemType, ItemDefinition] = {
"radio_station": ItemDefinition(
default_title="radio",
capabilities=("editable", "carryable", "deletable", "usable"),
use_sound=None,
emit_sound=None,
default_params={"streamUrl": "", "enabled": True, "channel": "stereo", "volume": 50, "effect": "off", "effectValue": 50},
),
"dice": ItemDefinition(
default_title="Dice",
capabilities=("editable", "carryable", "deletable", "usable"),
use_sound="sounds/roll.ogg",
emit_sound="sounds/roll.ogg",
default_params={"sides": 6, "number": 2},
),
"wheel": ItemDefinition(
default_title="wheel",
capabilities=("editable", "carryable", "deletable", "usable"),
use_sound="sounds/spin.ogg",
emit_sound="sounds/spin.ogg",
default_params={"spaces": "yes, no"},
use_cooldown_ms=4000,
),
"clock": ItemDefinition(
default_title="clock",
capabilities=("editable", "carryable", "deletable", "usable"),
emit_sound="sounds/clock.ogg",
default_params={"timeZone": CLOCK_DEFAULT_TIME_ZONE, "use24Hour": False},
),
}

6
server/app/item_service.py
View File

@@ -25,7 +25,7 @@ class ItemService:
def now_ms() -> int:
return int(time.time() * 1000)
def default_item(self, client: ClientConnection, item_type: Literal["radio_station", "dice", "wheel"]) -> WorldItem:
def default_item(self, client: ClientConnection, item_type: Literal["radio_station", "dice", "wheel", "clock"]) -> WorldItem:
item_def = get_item_definition(item_type)
now = self.now_ms()
return WorldItem(
@@ -39,7 +39,7 @@ class ItemService:
updatedAt=now,
version=1,
capabilities=list(item_def.capabilities),
useSound=item_def.use_sound,
emitSound=item_def.emit_sound,
params=deepcopy(item_def.default_params),
carrierId=None,
)
@@ -95,7 +95,7 @@ class ItemService:
updatedAt=persisted.updatedAt,
version=persisted.version,
capabilities=list(item_def.capabilities),
useSound=item_def.use_sound,
emitSound=item_def.emit_sound,
params=persisted.params,
carrierId=persisted.carrierId,
)

8
server/app/models.py
View File

@@ -40,7 +40,7 @@ class PingPacket(BasePacket):
class ItemAddPacket(BasePacket):
type: Literal["item_add"]
itemType: Literal["radio_station", "dice", "wheel"]
itemType: Literal["radio_station", "dice", "wheel", "clock"]
class ItemPickupPacket(BasePacket):
@@ -152,7 +152,7 @@ class NicknameResultPacket(BasePacket):
class WorldItem(BaseModel):
id: str
type: Literal["radio_station", "dice", "wheel"]
type: Literal["radio_station", "dice", "wheel", "clock"]
title: str
x: int
y: int
@@ -161,7 +161,7 @@ class WorldItem(BaseModel):
updatedAt: int
version: int
capabilities: list[str]
useSound: str | None = None
emitSound: str | None = None
params: dict
carrierId: str | None = None
@@ -169,7 +169,7 @@ class WorldItem(BaseModel):
class PersistedWorldItem(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
type: Literal["radio_station", "dice", "wheel"]
type: Literal["radio_station", "dice", "wheel", "clock"]
title: str
x: int
y: int

72
server/app/server.py
View File

@@ -2,6 +2,7 @@ from __future__ import annotations
import argparse
import asyncio
from datetime import datetime
import json
import logging
import random
@@ -9,13 +10,14 @@ import ssl
import uuid
from pathlib import Path
from typing import Literal
from zoneinfo import ZoneInfo
from pydantic import ValidationError, TypeAdapter
from websockets.asyncio.server import ServerConnection, serve
from .client import ClientConnection
from .config import load_config
from .item_catalog import get_item_use_cooldown_ms
from .item_catalog import CLOCK_DEFAULT_TIME_ZONE, CLOCK_TIME_ZONE_OPTIONS, get_item_use_cooldown_ms
from .item_service import ItemService
from .models import (
BroadcastChatMessagePacket,
@@ -90,6 +92,39 @@ class SignalingServer:
def _item_type_label(item: WorldItem) -> str:
return "radio" if item.type == "radio_station" else item.type
@staticmethod
def _normalize_clock_timezone(value: object) -> str:
token = str(value or "").strip()
if token in CLOCK_TIME_ZONE_OPTIONS:
return token
return CLOCK_DEFAULT_TIME_ZONE
@staticmethod
def _parse_clock_use_24_hour(value: object) -> bool | None:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
if isinstance(value, str):
token = value.strip().lower()
if token in {"on", "true", "1", "yes"}:
return True
if token in {"off", "false", "0", "no"}:
return False
return None
@classmethod
def _format_clock_display_time(cls, params: dict) -> str:
tz_name = cls._normalize_clock_timezone(params.get("timeZone"))
use_24_hour = cls._parse_clock_use_24_hour(params.get("use24Hour"))
if use_24_hour is None:
use_24_hour = False
now = datetime.now(ZoneInfo(tz_name))
if use_24_hour:
return now.strftime("%H:%M")
hour_12 = now.hour % 12 or 12
return f"{hour_12}:{now.minute:02d} {'AM' if now.hour < 12 else 'PM'}"
async def _send_item_result(
self,
client: ClientConnection,
@@ -432,7 +467,7 @@ class SignalingServer:
if item.carrierId is None and (item.x != client.x or item.y != client.y):
await self._send_item_result(client, False, "use", "Item is not on your square.", item.id)
return
if item.type not in {"radio_station", "dice", "wheel"}:
if item.type not in {"radio_station", "dice", "wheel", "clock"}:
await self._send_item_result(client, False, "use", "This item cannot be used yet.", item.id)
return
now_ms = self.item_service.now_ms()
@@ -483,7 +518,7 @@ class SignalingServer:
f"{client.nickname} rolled {item.title}: {', '.join(str(value) for value in rolls)} (total {total})."
)
self_message = f"You rolled {item.title}: {', '.join(str(value) for value in rolls)} (total {total})."
else:
elif item.type == "wheel":
spaces_raw = item.params.get("spaces", "")
if isinstance(spaces_raw, str):
spaces = [token.strip() for token in spaces_raw.split(",") if token.strip()]
@@ -505,16 +540,20 @@ class SignalingServer:
self_message = f"You spin {item.title}."
delayed_wheel_self_result = str(landed)
delayed_wheel_others_result = str(landed)
else:
display_time = self._format_clock_display_time(item.params)
others_message = f"{client.nickname} checks {item.title}. {item.title} says {display_time}."
self_message = f"{item.title} says {display_time}."
await self._broadcast(
BroadcastChatMessagePacket(type="chat_message", message=others_message, system=True),
exclude=client.websocket,
)
if item.useSound:
if item.emitSound:
await self._broadcast(
ItemUseSoundPacket(
type="item_use_sound",
itemId=item.id,
sound=item.useSound,
sound=item.emitSound,
x=item.x,
y=item.y,
)
@@ -665,6 +704,29 @@ class SignalingServer:
)
return
next_params["effectValue"] = round(effect_value, 1)
if item.type == "clock":
time_zone = str(next_params.get("timeZone", CLOCK_DEFAULT_TIME_ZONE)).strip()
if time_zone not in CLOCK_TIME_ZONE_OPTIONS:
await self._send_item_result(
client,
False,
"update",
f"timeZone must be one of {', '.join(CLOCK_TIME_ZONE_OPTIONS)}.",
item.id,
)
return
use_24_hour = self._parse_clock_use_24_hour(next_params.get("use24Hour"))
if use_24_hour is None:
await self._send_item_result(
client,
False,
"update",
"use24Hour must be on/off.",
item.id,
)
return
next_params["timeZone"] = time_zone
next_params["use24Hour"] = use_24_hour
item.params = next_params
item.updatedAt = self.item_service.now_ms()
item.version += 1

4
server/tests/test_item_persistence.py
View File

@@ -27,9 +27,9 @@ def test_item_persistence_omits_global_type_properties(tmp_path: Path) -> None:
assert isinstance(saved, list)
assert len(saved) == 1
assert "capabilities" not in saved[0]
assert "useSound" not in saved[0]
assert "emitSound" not in saved[0]
reloaded = ItemService(state_file=state_file)
loaded_item = reloaded.items[item.id]
assert loaded_item.useSound == "sounds/roll.ogg"
assert loaded_item.emitSound == "sounds/roll.ogg"
assert "usable" in loaded_item.capabilities

65
server/tests/test_item_use_cooldown.py
View File

@@ -117,3 +117,68 @@ async def test_radio_channel_update_validates(monkeypatch: pytest.MonkeyPatch) -
)
assert send_payloads[-1].ok is False
assert "channel must be one of" in send_payloads[-1].message.lower()
@pytest.mark.asyncio
async def test_clock_use_reports_time_and_emits_sound(monkeypatch: pytest.MonkeyPatch) -> None:
server = SignalingServer("127.0.0.1", 8765, None, None)
ws = _fake_ws()
client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
server.clients[ws] = client
item = server.item_service.default_item(client, "clock")
server.item_service.add_item(item)
send_payloads: list[object] = []
broadcast_payloads: list[object] = []
async def fake_send(websocket: ServerConnection, packet: object) -> None:
send_payloads.append(packet)
async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
broadcast_payloads.append(packet)
monkeypatch.setattr(server, "_send", fake_send)
monkeypatch.setattr(server, "_broadcast", fake_broadcast)
monkeypatch.setattr(server.item_service, "now_ms", lambda: 30_000)
monkeypatch.setattr(server, "_format_clock_display_time", lambda _params: "2:15 PM")
await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
assert send_payloads[-1].ok is True
assert send_payloads[-1].message == f"{item.title} says 2:15 PM."
assert any(getattr(packet, "type", "") == "item_use_sound" for packet in broadcast_payloads)
@pytest.mark.asyncio
async def test_clock_timezone_update_validates(monkeypatch: pytest.MonkeyPatch) -> None:
server = SignalingServer("127.0.0.1", 8765, None, None)
ws = _fake_ws()
client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
server.clients[ws] = client
item = server.item_service.default_item(client, "clock")
server.item_service.add_item(item)
send_payloads: list[object] = []
async def fake_send(websocket: ServerConnection, packet: object) -> None:
send_payloads.append(packet)
async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
return
monkeypatch.setattr(server, "_send", fake_send)
monkeypatch.setattr(server, "_broadcast", fake_broadcast)
await server._handle_message(
client,
json.dumps({"type": "item_update", "itemId": item.id, "params": {"timeZone": "America/New_York"}}),
)
assert send_payloads[-1].ok is True
assert item.params.get("timeZone") == "America/New_York"
await server._handle_message(
client,
json.dumps({"type": "item_update", "itemId": item.id, "params": {"timeZone": "Invalid/Zone"}}),
)
assert send_payloads[-1].ok is False
assert "timezone must be one of" in send_payloads[-1].message.lower()