from __future__ import annotations
import time
from typing import Any, Dict
import requests
from device.base import BaseDevice
from state import save
class ButtonDevice(BaseDevice):
def __init__(self, state):
super().__init__(state)
self._ensure_channels(self._channel_count())
for ch in self.state.channels:
ch["state"] = "idle"
def _channel_count(self) -> int:
return 4
def status(self) -> Dict[str, Any]:
return {"channels": self.state.channels}
def action(self, action_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
if action_name == "simulate_click":
channel = params.get("channel", 0)
return self.trigger_click(channel)
return {
"status": "error",
"error": "IllegalActionOrParams",
"message": "Device does not support this action or params",
}
def trigger_click(self, channel: int) -> Dict[str, Any]:
if channel < 0 or channel >= len(self.state.channels):
return {
"status": "error",
"error": "IllegalActionOrParams",
"message": "Invalid channel index",
}
now = time.strftime("%Y-%m-%d %H:%M:%S")
self.state.channels[channel]["last_event"] = "click"
self.state.channels[channel]["last_event_time"] = now
save(self.state)
# Send event to server asynchronously (best-effort)
self._send_event("click", channel)
return {"status": "ok", "message": f"Click triggered on channel {channel}"}
def _send_event(self, event_name: str, channel: int) -> None:
server_url = self.state.server_url.rstrip("/")
url = f"{server_url}/events/new"
payload = {
"event_name": event_name,
"device_id": self.state.device_id,
"data": {"channel": channel},
}
try:
requests.post(url, json=payload, timeout=3)
except Exception:
pass # Best-effort; server may be offline