feat: cleanup activity construction code (#136)

yalexs/activity.py

@@ -155,7 +155,7 @@
     ACTION_LOCK_MANUAL_UNLOCK,
 }
 
-ACTIVITY_ACTION_STATES = {
+ACTIVITY_ACTION_STATES: dict[str, LockStatus | LockDoorStatus] = {
     ACTION_RF_SECURE: LockStatus.LOCKED,
     ACTION_RF_LOCK: LockStatus.LOCKED,
     ACTION_RF_UNLATCH: LockStatus.UNLATCHED,
@@ -207,7 +207,17 @@ class Source(StrEnum):
 
 # If we get a lock operation activity with the same time stamp as a moving
 # activity we want to use the non-moving activity since its the completed state.
-MOVING_STATES = {LockStatus.UNLOCKING, LockStatus.UNLATCHING, LockStatus.LOCKING}
+MOVING_STATES: set[LockStatus | LockDoorStatus] = {
+    LockStatus.UNLOCKING,
+    LockStatus.UNLATCHING,
+    LockStatus.LOCKING,
+}
+
+ACTIVITY_MOVING_STATES = {
+    action
+    for action, action_state in ACTIVITY_ACTION_STATES.items()
+    if action_state in MOVING_STATES
+}
 
 
 class ActivityType(Enum):
@@ -224,15 +234,10 @@ class ActivityType(Enum):
 class Activity:
     """Base class for activities."""
 
-    def __init__(
-        self, source: str, activity_type: ActivityType, data: dict[str, Any]
-    ) -> None:
+    def __init__(self, source: str, data: dict[str, Any]) -> None:
         """Initialize activity."""
         self._source = source
-        self._activity_type = activity_type
         self._data = data
-        self._entities: dict[str, Any] = data.get("entities", {})
-        self._info: dict[str, Any] = data.get("info", {})
 
     def __repr__(self):
         """Return the representation."""
@@ -242,6 +247,16 @@ def __repr__(self):
             f"device_name={self.device_name}>"
         )
 
+    @cached_property
+    def _entities(self) -> dict[str, Any]:
+        """Return the entities of the activity."""
+        return self._data.get("entities", {})
+
+    @cached_property
+    def _info(self) -> dict[str, Any]:
+        """Return the info of the activity."""
+        return self._data.get("info", {})
+
     @cached_property
     def was_pushed(self) -> bool:
         """Return if the activity was pushed."""
@@ -302,14 +317,6 @@ def device_type(self) -> str | None:
 class BaseDoorbellMotionActivity(Activity):
     """Base class for doorbell motion activities."""
 
-    def __init__(
-        self, source: str, activity_type: ActivityType, data: dict[str, Any]
-    ) -> None:
-        """Initialize doorbell motion activity."""
-        super().__init__(source, activity_type, data)
-        self._image: dict[str, Any] | None = self._info.get("image")
-        self._content_token = data.get("doorbell", {}).get("contentToken")
-
     def __repr__(self):
         return (
             f"<{self.__class__.__name__} action={self.action} activity_type={self.activity_type} "
@@ -319,6 +326,16 @@ def __repr__(self):
             f"content_token={self.content_token}"
         )
 
+    @cached_property
+    def _image(self) -> dict[str, Any] | None:
+        """Return the image of the activity."""
+        return self._info.get("image")
+
+    @cached_property
+    def _content_token(self) -> str | None:
+        """Return the content token of the activity."""
+        return self._data.get("doorbell", {}).get("contentToken")
+
     @cached_property
     def image_url(self):
         """Return the image URL of the activity."""
@@ -346,17 +363,13 @@ def image_created_at_datetime(self):
 class DoorbellMotionActivity(BaseDoorbellMotionActivity):
     """A motion activity."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize doorbell motion activity."""
-        super().__init__(source, ActivityType.DOORBELL_MOTION, data)
+    _activity_type = ActivityType.DOORBELL_MOTION
 
 
 class DoorbellImageCaptureActivity(BaseDoorbellMotionActivity):
     """A motion activity with an image."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize doorbell motion activity."""
-        super().__init__(source, ActivityType.DOORBELL_IMAGE_CAPTURE, data)
+    _activity_type = ActivityType.DOORBELL_IMAGE_CAPTURE
 
 
 class DoorbellBaseActionActivity(Activity):
@@ -385,25 +398,23 @@ def activity_end_time(self):
 class DoorbellDingActivity(DoorbellBaseActionActivity):
     """Doorbell ding activity."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize doorbell ding activity."""
-        super().__init__(source, ActivityType.DOORBELL_DING, data)
+    _activity_type = ActivityType.DOORBELL_DING
 
 
 class DoorbellViewActivity(DoorbellBaseActionActivity):
     """Doorbell view activity."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize doorbell view activity."""
-        super().__init__(source, ActivityType.DOORBELL_VIEW, data)
+    _activity_type = ActivityType.DOORBELL_VIEW
 
 
 class LockOperationActivity(Activity):
     """Lock operation activity."""
 
+    _activity_type = ActivityType.LOCK_OPERATION_WITHOUT_OPERATOR
+
     def __init__(self, source: str, data: dict[str, Any]) -> None:
         """Initialize lock operation activity."""
-        super().__init__(source, ActivityType.LOCK_OPERATION_WITHOUT_OPERATOR, data)
+        super().__init__(source, data)
         operated_by: str | None = None
         calling_user = self.calling_user
         first_name: str | None = calling_user.get("FirstName")
@@ -544,17 +555,13 @@ def operator_thumbnail_url(self):
 class DoorOperationActivity(Activity):
     """Door operation activity."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize door operation activity."""
-        super().__init__(source, ActivityType.DOOR_OPERATION, data)
+    _activity_type = ActivityType.DOOR_OPERATION
 
 
 class BridgeOperationActivity(Activity):
     """Bridge operation activity."""
 
-    def __init__(self, source: str, data: dict[str, Any]) -> None:
-        """Initialize bridge operation activity."""
-        super().__init__(source, ActivityType.BRIDGE_OPERATION, data)
+    _activity_type = ActivityType.BRIDGE_OPERATION
 
 
 ActivityTypes = Union[

yalexs/manager/activity.py

@@ -53,7 +53,9 @@ def __init__(
         self._august_gateway = august_gateway
         self._api = api
         self._house_ids = house_ids
-        self._latest_activities: dict[str, dict[ActivityType, Activity]] = {}
+        self._latest_activities: defaultdict[
+            str, dict[ActivityType, Activity | None]
+        ] = defaultdict(lambda: defaultdict(lambda: None))
         self._did_first_update = False
         self.pubnub = pubnub
         self._update_tasks: dict[str, asyncio.Task] = {}
@@ -211,10 +213,10 @@ def async_process_newer_device_activities(
         for activity in activities:
             device_id = activity.device_id
             activity_type = activity.activity_type
-            device_activities = latest_activities.setdefault(device_id, {})
+            device_activities = latest_activities[device_id]
             # Ignore activities that are older than the latest one unless it is a non
             # locking or unlocking activity with the exact same start time.
-            last_activity = device_activities.get(activity_type)
+            last_activity = device_activities[activity_type]
             # The activity stream can have duplicate activities. So we need
             # to call get_latest_activity to figure out if if the activity
             # is actually newer than the last one.

yalexs/util.py

@@ -2,13 +2,14 @@
 import random
 import ssl
 from functools import cache
-from typing import Optional, Union
+from typing import Optional, Union, TYPE_CHECKING
 
 from .activity import (
     ACTION_BRIDGE_OFFLINE,
     ACTION_BRIDGE_ONLINE,
     ACTIVITY_ACTION_STATES,
     MOVING_STATES,
+    ACTIVITY_MOVING_STATES,
     BridgeOperationActivity,
     DoorbellImageCaptureActivity,
     DoorbellMotionActivity,
@@ -18,35 +19,38 @@
 from .const import CONFIGURATION_URLS, Brand
 from .lock import LockDetail
 
+LockActivityTypes = Union[
+    LockOperationActivity, DoorOperationActivity, BridgeOperationActivity
+]
+DoorbellActivityTypes = Union[
+    DoorbellImageCaptureActivity, DoorbellMotionActivity, BridgeOperationActivity
+]
+
+if TYPE_CHECKING:
+    from .doorbell import DoorbellDetail
+
 
 def get_latest_activity(
-    activity1: Optional[
-        Union[LockOperationActivity, DoorOperationActivity, BridgeOperationActivity]
-    ],
-    activity2: Optional[
-        Union[LockOperationActivity, DoorOperationActivity, BridgeOperationActivity]
-    ],
-) -> Optional[
-    Union[LockOperationActivity, DoorOperationActivity, BridgeOperationActivity]
-]:
+    activity1: Optional[LockActivityTypes], activity2: Optional[LockActivityTypes]
+) -> Optional[LockActivityTypes]:
     """Return the latest activity."""
-    if not activity1:
-        return activity2
-    if not activity2:
-        return activity1
-    if (
-        activity1.activity_start_time <= activity2.activity_start_time
-        and ACTIVITY_ACTION_STATES.get(activity2.action) not in MOVING_STATES
-    ):
-        return activity2
-    return activity1
+    return (
+        activity2
+        if (
+            not activity1
+            or (
+                activity2
+                and activity2.action not in ACTIVITY_MOVING_STATES
+                and activity1.activity_start_time <= activity2.activity_start_time
+            )
+        )
+        else activity1
+    )
 
 
 def update_lock_detail_from_activity(
     lock_detail: LockDetail,
-    activity: Union[
-        LockOperationActivity, DoorOperationActivity, BridgeOperationActivity
-    ],
+    activity: LockActivityTypes,
 ) -> bool:
     """Update the LockDetail from an activity."""
     activity_end_time_utc = as_utc_from_local(activity.activity_end_time)
@@ -82,7 +86,9 @@ def update_lock_detail_from_activity(
     return True
 
 
-def update_doorbell_image_from_activity(doorbell_detail, activity):
+def update_doorbell_image_from_activity(
+    doorbell_detail: "DoorbellDetail", activity: DoorbellActivityTypes
+) -> bool:
     """Update the DoorDetail from an activity with a new image."""
     if activity.device_id != doorbell_detail.device_id:
         raise ValueError