Add clock alarm scheduling with formatted alarm time options
This commit is contained in:
@@ -4,7 +4,14 @@ from __future__ import annotations
|
||||
|
||||
LABEL = "clock"
|
||||
TOOLTIP = "It tells the time. What did you think it did?"
|
||||
EDITABLE_PROPERTIES: tuple[str, ...] = ("title", "timeZone", "use24Hour", "topOfHourAnnounce")
|
||||
EDITABLE_PROPERTIES: tuple[str, ...] = (
|
||||
"title",
|
||||
"timeZone",
|
||||
"use24Hour",
|
||||
"topOfHourAnnounce",
|
||||
"alarmEnabled",
|
||||
"alarmTime",
|
||||
)
|
||||
CAPABILITIES: tuple[str, ...] = ("editable", "carryable", "deletable", "usable")
|
||||
USE_SOUND: str | None = None
|
||||
EMIT_SOUND = "sounds/clock.ogg"
|
||||
@@ -55,12 +62,24 @@ TIME_ZONE_OPTIONS: tuple[str, ...] = (
|
||||
"Pacific/Pago_Pago",
|
||||
"UTC",
|
||||
)
|
||||
DEFAULT_PARAMS: dict = {"timeZone": DEFAULT_TIME_ZONE, "use24Hour": False, "topOfHourAnnounce": True}
|
||||
PARAM_KEYS: tuple[str, ...] = ("timeZone", "use24Hour", "topOfHourAnnounce")
|
||||
DEFAULT_PARAMS: dict = {
|
||||
"timeZone": DEFAULT_TIME_ZONE,
|
||||
"use24Hour": False,
|
||||
"topOfHourAnnounce": True,
|
||||
"alarmEnabled": False,
|
||||
"alarmTime": "",
|
||||
}
|
||||
PARAM_KEYS: tuple[str, ...] = ("timeZone", "use24Hour", "topOfHourAnnounce", "alarmEnabled", "alarmTime")
|
||||
|
||||
PROPERTY_METADATA: dict[str, dict[str, object]] = {
|
||||
"title": {"valueType": "text", "tooltip": "Display name spoken and shown for this item.", "maxLength": 80},
|
||||
"timeZone": {"valueType": "list", "tooltip": "Timezone used when the clock speaks time.", "options": list(TIME_ZONE_OPTIONS)},
|
||||
"use24Hour": {"valueType": "boolean", "tooltip": "Use 24 hour format instead of AM/PM."},
|
||||
"topOfHourAnnounce": {"valueType": "boolean", "tooltip": "Automatically announce time at the top of each hour."},
|
||||
"alarmEnabled": {"valueType": "boolean", "tooltip": "Enable one daily alarm announcement at the configured alarm time."},
|
||||
"alarmTime": {
|
||||
"valueType": "text",
|
||||
"tooltip": "Alarm time. Uses 24-hour HH:MM when 24 hour format is on, otherwise H:MM AM/PM.",
|
||||
"maxLength": 8,
|
||||
},
|
||||
}
|
||||
|
||||
52
server/app/items/types/clock/time_format.py
Normal file
52
server/app/items/types/clock/time_format.py
Normal file
@@ -0,0 +1,52 @@
|
||||
"""Clock alarm time parsing/format helpers shared by validation and runtime checks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
_TWELVE_HOUR_RE = re.compile(r"^(0?[1-9]|1[0-2]):([0-5]\d)\s*([AaPp][Mm])$")
|
||||
_TWENTY_FOUR_HOUR_RE = re.compile(r"^([01]?\d|2[0-3]):([0-5]\d)$")
|
||||
|
||||
|
||||
def parse_alarm_time_for_mode(value: object, use_24_hour: bool) -> tuple[int, int] | None:
|
||||
"""Parse alarm time using one explicit mode and return `(hour24, minute)`."""
|
||||
|
||||
raw = str(value or "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
if use_24_hour:
|
||||
match = _TWENTY_FOUR_HOUR_RE.fullmatch(raw)
|
||||
if not match:
|
||||
return None
|
||||
return int(match.group(1)), int(match.group(2))
|
||||
match = _TWELVE_HOUR_RE.fullmatch(raw)
|
||||
if not match:
|
||||
return None
|
||||
hour = int(match.group(1))
|
||||
minute = int(match.group(2))
|
||||
meridiem = match.group(3).upper()
|
||||
if meridiem == "AM":
|
||||
hour24 = 0 if hour == 12 else hour
|
||||
else:
|
||||
hour24 = 12 if hour == 12 else hour + 12
|
||||
return hour24, minute
|
||||
|
||||
|
||||
def parse_alarm_time_flexible(value: object) -> tuple[int, int] | None:
|
||||
"""Parse alarm time as either 12-hour or 24-hour format and return `(hour24, minute)`."""
|
||||
|
||||
parsed = parse_alarm_time_for_mode(value, use_24_hour=True)
|
||||
if parsed is not None:
|
||||
return parsed
|
||||
return parse_alarm_time_for_mode(value, use_24_hour=False)
|
||||
|
||||
|
||||
def format_alarm_time_for_mode(hour24: int, minute: int, use_24_hour: bool) -> str:
|
||||
"""Format one parsed alarm time tuple as canonical 12-hour or 24-hour text."""
|
||||
|
||||
if use_24_hour:
|
||||
return f"{hour24:02d}:{minute:02d}"
|
||||
meridiem = "AM" if hour24 < 12 else "PM"
|
||||
hour12 = hour24 % 12 or 12
|
||||
return f"{hour12}:{minute:02d} {meridiem}"
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
from ....models import WorldItem
|
||||
from ...helpers import keep_only_known_params, parse_bool_like_or_none
|
||||
from .definition import DEFAULT_TIME_ZONE, PARAM_KEYS, TIME_ZONE_OPTIONS
|
||||
from .time_format import format_alarm_time_for_mode, parse_alarm_time_flexible
|
||||
|
||||
|
||||
def validate_update(_item: WorldItem, next_params: dict) -> dict:
|
||||
@@ -19,7 +20,20 @@ def validate_update(_item: WorldItem, next_params: dict) -> dict:
|
||||
top_of_hour_announce = parse_bool_like_or_none(next_params.get("topOfHourAnnounce"))
|
||||
if top_of_hour_announce is None:
|
||||
raise ValueError("topOfHourAnnounce must be on/off.")
|
||||
alarm_enabled = parse_bool_like_or_none(next_params.get("alarmEnabled"))
|
||||
if alarm_enabled is None:
|
||||
raise ValueError("alarmEnabled must be on/off.")
|
||||
alarm_time_raw = str(next_params.get("alarmTime", "") or "").strip()
|
||||
parsed_alarm = parse_alarm_time_flexible(alarm_time_raw) if alarm_time_raw else None
|
||||
if alarm_enabled and parsed_alarm is None:
|
||||
raise ValueError("alarmTime must be a valid time (HH:MM or H:MM AM/PM) when alarm is on.")
|
||||
if alarm_time_raw and parsed_alarm is None:
|
||||
raise ValueError("alarmTime must be a valid time (HH:MM or H:MM AM/PM).")
|
||||
next_params["timeZone"] = time_zone
|
||||
next_params["use24Hour"] = use_24_hour
|
||||
next_params["topOfHourAnnounce"] = top_of_hour_announce
|
||||
next_params["alarmEnabled"] = alarm_enabled
|
||||
next_params["alarmTime"] = (
|
||||
format_alarm_time_for_mode(parsed_alarm[0], parsed_alarm[1], use_24_hour) if parsed_alarm is not None else ""
|
||||
)
|
||||
return keep_only_known_params(next_params, PARAM_KEYS)
|
||||
|
||||
@@ -44,6 +44,7 @@ from .item_catalog import (
|
||||
)
|
||||
from .item_type_handlers import get_item_type_handler
|
||||
from .item_service import ItemService
|
||||
from .items.types.clock.time_format import parse_alarm_time_flexible
|
||||
from .models import (
|
||||
AuthLoginPacket,
|
||||
AuthLogoutPacket,
|
||||
@@ -164,6 +165,7 @@ class SignalingServer:
|
||||
self._radio_metadata_task: asyncio.Task[None] | None = None
|
||||
self._clock_announce_task: asyncio.Task[None] | None = None
|
||||
self._clock_top_of_hour_markers: dict[str, str] = {}
|
||||
self._clock_alarm_markers: dict[str, str] = {}
|
||||
|
||||
@staticmethod
|
||||
def _resolve_server_version() -> str:
|
||||
@@ -487,8 +489,8 @@ class SignalingServer:
|
||||
return
|
||||
|
||||
@classmethod
|
||||
def _build_clock_announcement_sounds(cls, params: dict, *, top_of_hour: bool) -> list[str]:
|
||||
"""Build ordered EL640 sample URLs for one clock announcement."""
|
||||
def _build_clock_time_sounds(cls, params: dict) -> list[str]:
|
||||
"""Build ordered EL640 sample URLs for just the clock time phrase."""
|
||||
|
||||
tz_name = cls._normalize_clock_timezone(params.get("timeZone"))
|
||||
use_24_hour = cls._parse_clock_use_24_hour(params.get("use24Hour")) is True
|
||||
@@ -498,10 +500,7 @@ class SignalingServer:
|
||||
ampm = "AM" if hour24 < 12 else "PM"
|
||||
hour12 = hour24 % 12 or 12
|
||||
|
||||
sounds: list[str] = []
|
||||
if top_of_hour:
|
||||
sounds.append("/sounds/clock/el640/hour1.ogg")
|
||||
sounds.append("/sounds/clock/el640/its.ogg")
|
||||
sounds: list[str] = ["/sounds/clock/el640/its.ogg"]
|
||||
|
||||
if use_24_hour:
|
||||
if hour24 < 20:
|
||||
@@ -529,16 +528,30 @@ class SignalingServer:
|
||||
|
||||
if not use_24_hour:
|
||||
sounds.append(f"/sounds/clock/el640/{ampm}.ogg")
|
||||
if top_of_hour:
|
||||
return sounds
|
||||
|
||||
@classmethod
|
||||
def _build_clock_announcement_sounds(cls, params: dict, *, top_of_hour: bool, alarm: bool) -> list[str]:
|
||||
"""Build ordered EL640 sample URLs for one clock announcement variant."""
|
||||
|
||||
sounds: list[str] = []
|
||||
if alarm:
|
||||
sounds.append("/sounds/clock/el640/announcement.ogg")
|
||||
elif top_of_hour:
|
||||
sounds.append("/sounds/clock/el640/hour1.ogg")
|
||||
sounds.extend(cls._build_clock_time_sounds(params))
|
||||
if alarm:
|
||||
sounds.append("/sounds/clock/el640/alarm.ogg")
|
||||
elif top_of_hour:
|
||||
sounds.append("/sounds/clock/el640/hour2.ogg")
|
||||
return sounds
|
||||
|
||||
async def _broadcast_clock_announcement(self, item: WorldItem, *, top_of_hour: bool) -> None:
|
||||
async def _broadcast_clock_announcement(self, item: WorldItem, *, top_of_hour: bool, alarm: bool) -> None:
|
||||
"""Broadcast one server-authoritative clock speech sequence from item position."""
|
||||
|
||||
sound_x, sound_y = self._get_item_sound_source_position(item)
|
||||
sound_range = self._get_item_emit_range(item)
|
||||
sounds = self._build_clock_announcement_sounds(item.params, top_of_hour=top_of_hour)
|
||||
sounds = self._build_clock_announcement_sounds(item.params, top_of_hour=top_of_hour, alarm=alarm)
|
||||
if not sounds:
|
||||
return
|
||||
await self._broadcast(
|
||||
@@ -561,21 +574,30 @@ class SignalingServer:
|
||||
for stale_id in list(self._clock_top_of_hour_markers.keys()):
|
||||
if stale_id not in valid_clock_ids:
|
||||
self._clock_top_of_hour_markers.pop(stale_id, None)
|
||||
for stale_id in list(self._clock_alarm_markers.keys()):
|
||||
if stale_id not in valid_clock_ids:
|
||||
self._clock_alarm_markers.pop(stale_id, None)
|
||||
for item in self.items.values():
|
||||
if item.type != "clock":
|
||||
continue
|
||||
enabled = item.params.get("topOfHourAnnounce", True)
|
||||
if enabled is not True:
|
||||
continue
|
||||
tz_name = self._normalize_clock_timezone(item.params.get("timeZone"))
|
||||
now = datetime.now(ZoneInfo(tz_name))
|
||||
if now.minute != 0 or now.second > 1:
|
||||
continue
|
||||
marker = now.strftime("%Y-%m-%d-%H")
|
||||
if self._clock_top_of_hour_markers.get(item.id) == marker:
|
||||
continue
|
||||
self._clock_top_of_hour_markers[item.id] = marker
|
||||
await self._broadcast_clock_announcement(item, top_of_hour=True)
|
||||
top_of_hour_enabled = item.params.get("topOfHourAnnounce", True) is True
|
||||
if top_of_hour_enabled and now.minute == 0 and now.second <= 1:
|
||||
marker = now.strftime("%Y-%m-%d-%H")
|
||||
if self._clock_top_of_hour_markers.get(item.id) != marker:
|
||||
self._clock_top_of_hour_markers[item.id] = marker
|
||||
await self._broadcast_clock_announcement(item, top_of_hour=True, alarm=False)
|
||||
|
||||
alarm_enabled = item.params.get("alarmEnabled", False) is True
|
||||
alarm_time = parse_alarm_time_flexible(item.params.get("alarmTime", ""))
|
||||
if alarm_enabled and alarm_time is not None:
|
||||
alarm_hour, alarm_minute = alarm_time
|
||||
if now.hour == alarm_hour and now.minute == alarm_minute and now.second <= 1:
|
||||
marker = now.strftime("%Y-%m-%d-%H-%M")
|
||||
if self._clock_alarm_markers.get(item.id) != marker:
|
||||
self._clock_alarm_markers[item.id] = marker
|
||||
await self._broadcast_clock_announcement(item, top_of_hour=False, alarm=True)
|
||||
await asyncio.sleep(CLOCK_ANNOUNCE_POLL_INTERVAL_S)
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
@@ -1826,7 +1848,7 @@ class SignalingServer:
|
||||
)
|
||||
)
|
||||
if item.type == "clock":
|
||||
await self._broadcast_clock_announcement(item, top_of_hour=False)
|
||||
await self._broadcast_clock_announcement(item, top_of_hour=False, alarm=False)
|
||||
if item.type == "piano":
|
||||
await self._send_piano_status(
|
||||
client,
|
||||
|
||||
@@ -282,6 +282,29 @@ async def test_clock_timezone_update_validates(monkeypatch: pytest.MonkeyPatch)
|
||||
assert send_payloads[-1].ok is False
|
||||
assert "timezone must be one of" in send_payloads[-1].message.lower()
|
||||
|
||||
await server._handle_message(
|
||||
client,
|
||||
json.dumps({"type": "item_update", "itemId": item.id, "params": {"alarmEnabled": True}}),
|
||||
)
|
||||
assert send_payloads[-1].ok is False
|
||||
assert "alarmtime must be a valid time" in send_payloads[-1].message.lower()
|
||||
|
||||
await server._handle_message(
|
||||
client,
|
||||
json.dumps({"type": "item_update", "itemId": item.id, "params": {"alarmTime": "3:15 PM", "alarmEnabled": True}}),
|
||||
)
|
||||
assert send_payloads[-1].ok is True
|
||||
assert item.params.get("alarmEnabled") is True
|
||||
assert item.params.get("alarmTime") == "3:15 PM"
|
||||
|
||||
await server._handle_message(
|
||||
client,
|
||||
json.dumps({"type": "item_update", "itemId": item.id, "params": {"use24Hour": True}}),
|
||||
)
|
||||
assert send_payloads[-1].ok is True
|
||||
assert item.params.get("use24Hour") is True
|
||||
assert item.params.get("alarmTime") == "15:15"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_wheel_use_does_not_consume_cooldown(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
|
||||
@@ -167,6 +167,21 @@ async def test_item_secondary_use_missing_handler_returns_message(monkeypatch: p
|
||||
assert "No secondary action" in results[-1].message
|
||||
|
||||
|
||||
def test_clock_alarm_announcement_sequence_shape() -> None:
|
||||
server = SignalingServer("127.0.0.1", 8765, None, None, grid_size=41)
|
||||
params = {"timeZone": "America/Detroit", "use24Hour": False}
|
||||
|
||||
alarm_sounds = server._build_clock_announcement_sounds(params, top_of_hour=False, alarm=True)
|
||||
assert alarm_sounds
|
||||
assert alarm_sounds[0] == "/sounds/clock/el640/announcement.ogg"
|
||||
assert alarm_sounds[-1] == "/sounds/clock/el640/alarm.ogg"
|
||||
|
||||
top_of_hour_sounds = server._build_clock_announcement_sounds(params, top_of_hour=True, alarm=False)
|
||||
assert top_of_hour_sounds
|
||||
assert top_of_hour_sounds[0] == "/sounds/clock/el640/hour1.ogg"
|
||||
assert top_of_hour_sounds[-1] == "/sounds/clock/el640/hour2.ogg"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_login_uses_hash_offload(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||
|
||||
Reference in New Issue
Block a user