successful pydantic experiments

apps/cubes/cube.py

@@ -1,31 +1,83 @@
 import json
 import logging
 import sys
+from enum import Enum
 from pathlib import Path
+from typing import Annotated, Literal, Optional, TypedDict
 
 from appdaemon.plugins.hass.hassapi import Hass
 from appdaemon.plugins.mqtt.mqttapi import Mqtt
+from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 rc_path = (Path(__file__).resolve().parents[1] / 'room_control').as_posix()
 sys.path.insert(0, rc_path)
 from console import console, setup_component_logging
 
 
+class Side(int, Enum):
+    back = 0
+    right = 1
+    top = 2
+    front = 3
+    left = 4
+    bottom = 5
+    
+
+Actions = Literal[
+    '', 'wakeup',
+    'slide', 'shake',
+    'rotate_left', 'rotate_right',
+    'flip180', 'flip90'
+]
+
+
+class CubeEvent(BaseModel):
+    action: Optional[Actions] = None
+    side: Optional[Side] = Field(default=None, ge=0)
+    action_angle: Optional[float] = None
+    action_side: Optional[Side] = None
+    action_from_side: Optional[Side] = None
+    action_to_side: Optional[Side] = None
+    battery: Optional[int] = Field(default=None, ge=0)
+    current: Optional[int] = Field(default=None, ge=0)
+    device_temperature: Optional[int] = Field(default=None, ge=0)
+    linkquality: Optional[int] = Field(default=None, ge=0)
+    power: Optional[int] = Field(default=None, ge=0)
+    power_outage_count: Optional[int] = Field(default=None, ge=0)
+    voltage: Optional[int] = Field(default=None, ge=0)
+
+
+class MQTTResponse(BaseModel):
+    topic: str
+    payload: CubeEvent
+
+    @field_validator('payload', mode='before')
+    def payload_str(cls, v: str):
+        return CubeEvent.model_validate_json(v)
+
+
+class CallbackEntry(BaseModel):
+    entity: str
+    event: Optional[str] = None
+    type: str
+    kwargs: str
+    function: str
+    name: str
+    pin_app: bool
+    pin_thread: int
+
+
+class Callbacks(TypedDict):
+    pass
+
+
+Callbacks = dict[str, dict[str, CallbackEntry]]
+
+
 class AqaraCube(Hass, Mqtt):
     def initialize(self):
         setup_component_logging(self)
         self.logger.setLevel(self.args.get('rich', logging.INFO))
-        # self.logger = logging.getLogger()
-
-        # if (level := self.args.get('rich', False)):
-        #     init_logging(self, level)
-
-        # for name, logger in logging.Logger.manager.loggerDict.items():
-        #     logger = logging.getLogger(name)
-        #     if logger.hasHandlers():
-        #         self.log(f'[yellow]{name.ljust(25)}[/]{type(logger).__name__}', )
-        #         for handler in logger.handlers:
-        #             self.log(f'  {type(handler).__name__}: {type(handler.formatter).__name__}')
 
         topic = f'zigbee2mqtt/{self.args["cube"]}'
         self.mqtt_subscribe(topic, namespace='mqtt')
@@ -35,22 +87,33 @@ class AqaraCube(Hass, Mqtt):
         self.app = self.get_app(self.args['app'])
         self.log(f'Connected to AD app: [room]{self.app.name}[/]')
 
+        self.log(self.callbacks())
+
     def terminate(self):
         self.log('[bold red]Terminating[/]', level='DEBUG')
 
+    def callbacks(self) -> dict[str, CallbackEntry]:
+        data = TypeAdapter(Callbacks).validate_python(self.get_callback_entries())
+        try:
+            return data[self.name]
+        except KeyError:
+            return []
+
     def handle_event(self, event_name, data, cb_args):
-        payload = json.loads(data['payload'])
+        data = MQTTResponse.model_validate(data)
-        action = payload['action']
+        action = data.payload.action
 
         if action == '' or action == 'wakeup':
             return
         else:
-            log_msg = f'{event_name} on [topic]{data["topic"]}[/], Action: "[yellow]{str(action)}[/]"\n'
+            self.log(
-            self.log(log_msg + json.dumps(payload, indent=2), level='DEBUG')
+                f'{event_name} on [topic]{data.topic}[/], Action: "[yellow]{str(action)}[/]"',
+                level='DEBUG'
+            )
             if (arg := self.args.get(action, False)):
                 self.action_handler(action=action, description=arg)
             elif handler := getattr(self, f'handle_{action}', None):
-                handler(payload)
+                handler(data.payload)
 
     def action_handler(self, action: str, description: str):
         self.log(f'{self.args["cube"]}: {action}: {description}')
@@ -68,21 +131,3 @@ class AqaraCube(Hass, Mqtt):
 
         else:
             self.log(f'Unhandled action: {action}', level='WARNING')
-
-    # def handle_rotate_right(self, payload):
-    #     self.log(f'{self.args["cube"]}: Rotate right')
-
-    # def handle_rotate_left(self, payload):
-    #     self.log(f'{self.args["cube"]}: Rotate left')
-
-    # def handle_slide(self, payload):
-    #     self.log(f'{self.args["cube"]}: Slide')
-
-    # def handle_flip180(self, payload):
-    #     self.log(f'{self.args["cube"]}: Flipped 180')
-
-    # def handle_flip90(self, payload):
-    #     self.log(f'{self.args["cube"]}: Flipped 90')
-
-    # def handle_shake(self, payload):
-    #     self.log(f'{self.args["cube"]}: Shake')

apps/scene_detect.py

@@ -27,7 +27,7 @@ class MotionCanceller(SceneDetector):
                 callback=lambda *args, **kwargs: app.cancel_motion_callback(),
                 delay=0.5
             )
-        except:
+        except Exception:
             self.log(f'Error cancelling motion callback for {self.args["app"]}', level='ERROR')
         else:
             self.log('Cancelled motion callback')