2024-11-01 15:36:09 +00:00
|
|
|
"""Implementation of trigger logs module."""
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
2024-11-20 13:21:08 +00:00
|
|
|
from dataclasses import dataclass
|
|
|
|
from typing import Annotated
|
2024-11-01 15:36:09 +00:00
|
|
|
|
2024-11-20 13:21:08 +00:00
|
|
|
from mashumaro import DataClassDictMixin
|
|
|
|
from mashumaro.types import Alias
|
2024-11-01 15:36:09 +00:00
|
|
|
|
|
|
|
from ..smartmodule import SmartModule
|
|
|
|
|
|
|
|
|
2024-11-20 13:21:08 +00:00
|
|
|
@dataclass
|
|
|
|
class LogEntry(DataClassDictMixin):
|
2024-11-01 15:36:09 +00:00
|
|
|
"""Presentation of a single log entry."""
|
|
|
|
|
|
|
|
id: int
|
2024-11-20 13:21:08 +00:00
|
|
|
event_id: Annotated[str, Alias("eventId")]
|
|
|
|
timestamp: int
|
2024-11-01 15:36:09 +00:00
|
|
|
event: str
|
|
|
|
|
|
|
|
|
|
|
|
class TriggerLogs(SmartModule):
|
|
|
|
"""Implementation of trigger logs."""
|
|
|
|
|
|
|
|
REQUIRED_COMPONENT = "trigger_log"
|
|
|
|
MINIMUM_UPDATE_INTERVAL_SECS = 60 * 60
|
|
|
|
|
|
|
|
def query(self) -> dict:
|
|
|
|
"""Query to execute during the update cycle."""
|
|
|
|
return {"get_trigger_logs": {"start_id": 0}}
|
|
|
|
|
|
|
|
@property
|
|
|
|
def logs(self) -> list[LogEntry]:
|
|
|
|
"""Return logs."""
|
2024-11-20 13:21:08 +00:00
|
|
|
return [LogEntry.from_dict(log) for log in self.data["logs"]]
|