import time from typing import Dict from services.runtime_state import get_state, set_state # category -> unix timestamp until muted def _mutes() -> Dict[str, float]: return get_state().get("mutes", {}) def _save(mutes: Dict[str, float]): set_state("mutes", mutes) def _cleanup() -> None: mutes = _mutes() now = time.time() expired = [k for k, until in mutes.items() if until <= now] for k in expired: mutes.pop(k, None) _save(mutes) def set_mute(category: str, seconds: int) -> float: _cleanup() mutes = _mutes() until = time.time() + max(0, seconds) mutes[category] = until _save(mutes) return until def clear_mute(category: str) -> None: mutes = _mutes() mutes.pop(category, None) _save(mutes) def is_muted(category: str | None) -> bool: if not category: return False _cleanup() mutes = _mutes() until = mutes.get(category) if until is None: return False if until <= time.time(): mutes.pop(category, None) _save(mutes) return False return True def list_mutes() -> dict[str, int]: _cleanup() now = time.time() mutes = _mutes() return {k: int(until - now) for k, until in mutes.items()} def is_auto_muted(cfg: dict, category: str | None) -> bool: if not category: return False auto_list = cfg.get("alerts", {}).get("auto_mute", []) if not isinstance(auto_list, list): return False now = time.localtime() now_minutes = now.tm_hour * 60 + now.tm_min for item in auto_list: if not isinstance(item, dict): continue cat = item.get("category") if cat != category: continue start = item.get("start", "00:00") end = item.get("end", "00:00") try: sh, sm = [int(x) for x in start.split(":")] eh, em = [int(x) for x in end.split(":")] except Exception: continue start_min = sh * 60 + sm end_min = eh * 60 + em if start_min == end_min: continue if start_min < end_min: if start_min <= now_minutes < end_min: return True else: if now_minutes >= start_min or now_minutes < end_min: return True return False