feat(action-type): add new concept of action types to run different t…

…ype of actions

related to #173

Pipfile

@@ -9,6 +9,7 @@ pytest = "==6.2.1"
 pytest-asyncio = "==0.14.0"
 pytest-cov = "==2.10.1"
 pytest-mock = "==3.5.0"
+pytest-timeout = "==1.4.2"
 mock = "==4.0.3"
 pre-commit = "==2.9.3"
 commitizen = "==2.13.0"

apps/controllerx/cx_const.py

@@ -1,10 +1,15 @@
-from typing import Any, Awaitable, Callable, Dict, Tuple, Union
+from typing import Any, Awaitable, Callable, Dict, List, Mapping, Tuple, Union
 
 ActionFunction = Callable[..., Awaitable[Any]]
-TypeAction = Union[ActionFunction, Tuple[Any, ...], str]
+ActionFunctionWithParams = Tuple[ActionFunction, Tuple]
+TypeAction = Union[ActionFunction, ActionFunctionWithParams]
 ActionEvent = Union[str, int]
-TypeActionsMapping = Dict[ActionEvent, TypeAction]
-ActionsMapping = Dict[ActionEvent, Union[ActionFunction, Tuple[Any, ...]]]
+PredefinedActionsMapping = Dict[str, TypeAction]
+DefaultActionsMapping = Mapping[ActionEvent, str]
+
+CustomAction = Union[str, Dict[str, Any]]
+CustomActions = Union[List[CustomAction], CustomAction]
+CustomActionsMapping = Mapping[ActionEvent, CustomActions]
 
 
 class Light:

apps/controllerx/cx_core/action_type/__init__.py

@@ -0,0 +1,46 @@
+from typing import TYPE_CHECKING, Dict, List, Type
+
+from cx_const import ActionEvent, CustomAction, CustomActions
+from cx_core.action_type.base import ActionType
+from cx_core.action_type.call_service_action_type import CallServiceActionType
+from cx_core.action_type.delay_action_type import DelayActionType
+from cx_core.action_type.predefined_action_type import PredefinedActionType
+from cx_core.action_type.scene_action_type import SceneActionType
+
+if TYPE_CHECKING:
+    from cx_core import Controller
+
+ActionsMapping = Dict[ActionEvent, List[ActionType]]
+
+action_type_mapping: Dict[str, Type[ActionType]] = {
+    "action": PredefinedActionType,
+    "service": CallServiceActionType,
+    "scene": SceneActionType,
+    "delay": DelayActionType,
+}
+
+
+def parse_actions(controller: "Controller", data: CustomActions) -> List[ActionType]:
+    actions: CustomActions
+    if isinstance(data, (list, tuple)):
+        actions = list(data)
+    else:
+        actions = [data]
+
+    return [_parse_action(controller, action) for action in actions]
+
+
+def _parse_action(controller: "Controller", action: CustomAction) -> ActionType:
+    if isinstance(action, str):
+        return PredefinedActionType(controller, {"action": action})
+    try:
+        return next(
+            action_type(controller, action)
+            for key in action
+            for action_type_key, action_type in action_type_mapping.items()
+            if key == action_type_key
+        )
+    except StopIteration:
+        raise ValueError(
+            f"Not able to parse `{action}`. Available keys are: {action_type_mapping.keys()}"
+        )

apps/controllerx/cx_core/action_type/base.py

@@ -0,0 +1,25 @@
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+from cx_core.integration import EventData
+
+if TYPE_CHECKING:
+    from cx_core import Controller
+
+
+class ActionType(ABC):
+    controller: "Controller"
+
+    def __init__(self, controller: "Controller", action: Dict[str, Any]) -> None:
+        self.controller = controller
+        self.initialize(**action)
+
+    def initialize(self, **kwargs) -> None:
+        pass
+
+    @abstractmethod
+    async def run(self, extra: Optional[EventData] = None) -> None:
+        raise NotImplementedError
+
+    def __str__(self) -> str:
+        return f"{self.__class__.__name__}"

apps/controllerx/cx_core/action_type/call_service_action_type.py

@@ -0,0 +1,28 @@
+from typing import Any, Dict, Optional
+
+from cx_core.action_type.base import ActionType
+from cx_core.integration import EventData
+
+
+class CallServiceActionType(ActionType):
+    service: str
+    entity_id: Optional[str]
+    data: Dict[str, Any]
+
+    def initialize(self, **kwargs) -> None:
+        self.service = kwargs["service"]
+        self.entity_id = kwargs.get("entity_id")
+        self.data = kwargs.get("data", {})
+
+    async def run(self, extra: Optional[EventData] = None) -> None:
+        if self.entity_id:
+            if "entity_id" in self.data:
+                del self.data["entity_id"]
+            await self.controller.call_service(
+                self.service, entity_id=self.entity_id, **self.data
+            )
+        else:
+            await self.controller.call_service(self.service, **self.data)
+
+    def __str__(self) -> str:
+        return f"Service ({self.service})"

apps/controllerx/cx_core/action_type/delay_action_type.py

@@ -0,0 +1,17 @@
+from typing import Optional
+
+from cx_core.action_type.base import ActionType
+from cx_core.integration import EventData
+
+
+class DelayActionType(ActionType):
+    delay: int
+
+    def initialize(self, **kwargs) -> None:
+        self.delay = kwargs["delay"]
+
+    async def run(self, extra: Optional[EventData] = None) -> None:
+        await self.controller.sleep(self.delay)
+
+    def __str__(self) -> str:
+        return f"Delay ({self.delay} seconds)"

apps/controllerx/cx_core/action_type/predefined_action_type.py

@@ -0,0 +1,39 @@
+import inspect
+from typing import Optional
+
+from cx_const import ActionFunctionWithParams, PredefinedActionsMapping, TypeAction
+from cx_core.action_type.base import ActionType
+from cx_core.integration import EventData
+
+
+def _get_action(action_value: TypeAction) -> ActionFunctionWithParams:
+    if isinstance(action_value, tuple):
+        return action_value
+    else:
+        return (action_value, tuple())
+
+
+class PredefinedActionType(ActionType):
+    action_key: str
+    predefined_actions_mapping: PredefinedActionsMapping
+
+    def initialize(self, **kwargs) -> None:
+        self.action_key = kwargs["action"]
+        self.predefined_actions_mapping = (
+            self.controller.get_predefined_actions_mapping()
+        )
+        if self.action_key not in self.predefined_actions_mapping:
+            raise ValueError(
+                f"`{self.action_key}` is not one of the predefined actions."
+                f"Available actions are: {list(self.predefined_actions_mapping.keys())}"
+            )
+
+    async def run(self, extra: Optional[EventData] = None) -> None:
+        action, args = _get_action(self.predefined_actions_mapping[self.action_key])
+        if "extra" in set(inspect.signature(action).parameters):
+            await action(*args, extra=extra)
+        else:
+            await action(*args)
+
+    def __str__(self) -> str:
+        return f"Predefined ({self.action_key})"

apps/controllerx/cx_core/action_type/scene_action_type.py

@@ -0,0 +1,17 @@
+from typing import Optional
+
+from cx_core.action_type.base import ActionType
+from cx_core.integration import EventData
+
+
+class SceneActionType(ActionType):
+    scene: str
+
+    def initialize(self, **kwargs) -> None:
+        self.scene = kwargs["scene"]
+
+    async def run(self, extra: Optional[EventData] = None) -> None:
+        await self.controller.call_service("scene/turn_on", entity_id=self.scene)
+
+    def __str__(self) -> str:
+        return f"Scene ({self.scene})"

apps/controllerx/cx_core/color_helper.py

@@ -74,7 +74,7 @@ def get_color_wheel(colors: Union[str, Colors]) -> Colors:
                 f"`{colors}` is not an option for `color_wheel`. Options are: {list(COLOR_WHEELS.keys())}"
             )
         return COLOR_WHEELS[colors]
-    elif isinstance(colors, list):
+    elif isinstance(colors, (list, tuple)):
         return colors
     else:
         raise ValueError(