Initial trigger logs implementation (#900)

Co-authored-by: Steven B. <51370195+sdb9696@users.noreply.github.com>

kasa/cli/device.py

@@ -193,3 +193,13 @@ async def update_credentials(dev, username, password):
     click.confirm("Do you really want to replace the existing credentials?", abort=True)
 
     return await dev.update_credentials(username, password)
+
+
+@device.command(name="logs")
+@pass_dev_or_child
+async def child_logs(dev):
+    """Print child device trigger logs."""
+    if logs := dev.modules.get(Module.TriggerLogs):
+        await dev.update(update_children=True)
+        for entry in logs.logs:
+            print(entry)

kasa/module.py

@@ -127,6 +127,7 @@ class Module(ABC):
     WaterleakSensor: Final[ModuleName[smart.WaterleakSensor]] = ModuleName(
         "WaterleakSensor"
     )
+    TriggerLogs: Final[ModuleName[smart.TriggerLogs]] = ModuleName("TriggerLogs")
 
     # SMARTCAMERA only modules
     Camera: Final[ModuleName[experimental.Camera]] = ModuleName("Camera")

kasa/smart/modules/__init__.py

@@ -27,6 +27,7 @@ from .reportmode import ReportMode
 from .temperaturecontrol import TemperatureControl
 from .temperaturesensor import TemperatureSensor
 from .time import Time
+from .triggerlogs import TriggerLogs
 from .waterleaksensor import WaterleakSensor
 
 __all__ = [
@@ -56,6 +57,7 @@ __all__ = [
     "WaterleakSensor",
     "ContactSensor",
     "MotionSensor",
+    "TriggerLogs",
     "FrostProtection",
     "SmartLightEffect",
 ]

kasa/smart/modules/triggerlogs.py

@@ -0,0 +1,34 @@
+"""Implementation of trigger logs module."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic.v1 import BaseModel, Field, parse_obj_as
+
+from ..smartmodule import SmartModule
+
+
+class LogEntry(BaseModel):
+    """Presentation of a single log entry."""
+
+    id: int
+    event_id: str = Field(alias="eventId")
+    timestamp: datetime
+    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."""
+        return parse_obj_as(list[LogEntry], self.data["logs"])