Functions wrapped in @in_separate_thread now return the created Thread; The logging system has been redesigned, the global logger is now configured; EVENT_ROUTING_MAP is filled automatically; Requirements updated;

Author: gri-gus

README.md

@@ -46,6 +46,15 @@ or
 pip install streamdeck_sdk
 ```
 
+## Features
+
+* Easy use. You can quickly create your own plugin without having to understand how websockets and other complicated
+  things work.
+* Fully typed, using [pydantic](https://github.com/pydantic/pydantic).
+* Includes image to base64 converters for easy installation of icons on keys.
+* Includes a decorator for functions and methods to run on a separate thread.
+* Exception logging and easy logging configuration.
+
 ## Examples
 
 [LoremFlickr](https://github.com/gri-gus/loremflickr-streamdeck-plugin) - Plugin for installing images from LoremFlickr

setup.py

@@ -1,6 +1,6 @@
 import setuptools
 
-VERSION = "0.2.1"
+VERSION = "0.3.0"
 PACKAGE_DIR = "src"
 REQUIREMENTS_FILE = PACKAGE_DIR + "/requirements.txt"
 README = "README.md"

src/requirements.txt

@@ -1,4 +1,4 @@
-decohints>=1.0.7
-pydantic>=1.10.5
+decohints>=1.0.9
+pydantic>=1.10.7
 typing_extensions>=4.5.0
 websocket-client>=1.5.1

src/streamdeck_sdk/__init__.py

@@ -1,7 +1,5 @@
-from . import events_received_objs
-from . import events_sent_objs
 from . import mixins
-from . import registration_objs
+from .sd_objs import events_received_objs, events_sent_objs, registration_objs
 from .logger import logger, log_errors
 from .sdk import StreamDeck, Action
 from .utils import image_file_to_base64, image_bytes_to_base64, in_separate_thread

src/streamdeck_sdk/event_routings.py

@@ -1,108 +1,59 @@
-from dataclasses import dataclass
-from typing import Type, Dict
+from enum import Enum
+from typing import Type, Dict, get_type_hints
 
-import pydantic
+from pydantic import BaseModel
 
-from . import events_received_objs
+from . import mixins
 
 
-@dataclass
-class EventRoutingObj:
-    handler_name: str
-    obj: Type[pydantic.BaseModel]
-
+class EventRoutingObjTypes(Enum):
+    ACTION = "ACTION"
+    PLUGIN = "PLUGIN"
 
-ACTION_EVENT_ROUTING_MAP: Dict[str, EventRoutingObj] = {
-    "didReceiveSettings": EventRoutingObj(
-        obj=events_received_objs.DidReceiveSettings,
-        handler_name="on_did_receive_settings",
-    ),
-    "keyDown": EventRoutingObj(
-        obj=events_received_objs.KeyDown,
-        handler_name="on_key_down",
-    ),
-    "keyUp": EventRoutingObj(
-        obj=events_received_objs.KeyUp,
-        handler_name="on_key_up",
-    ),
-    "touchTap": EventRoutingObj(
-        obj=events_received_objs.TouchTap,
-        handler_name="on_touch_tap",
-    ),
-    "dialDown": EventRoutingObj(
-        obj=events_received_objs.DialDown,
-        handler_name="on_dial_down",
-    ),
-    "dialUp": EventRoutingObj(
-        obj=events_received_objs.DialUp,
-        handler_name="on_dial_up",
-    ),
-    "dialPress": EventRoutingObj(
-        obj=events_received_objs.DialPress,
-        handler_name="on_dial_press",
-    ),
-    "dialRotate": EventRoutingObj(
-        obj=events_received_objs.DialRotate,
-        handler_name="on_dial_rotate",
-    ),
-    "willAppear": EventRoutingObj(
-        obj=events_received_objs.WillAppear,
-        handler_name="on_will_appear",
-    ),
-    "willDisappear": EventRoutingObj(
-        obj=events_received_objs.WillDisappear,
-        handler_name="on_will_disappear",
-    ),
-    "titleParametersDidChange": EventRoutingObj(
-        obj=events_received_objs.TitleParametersDidChange,
-        handler_name="on_title_parameters_did_change",
-    ),
-    "propertyInspectorDidAppear": EventRoutingObj(
-        obj=events_received_objs.PropertyInspectorDidAppear,
-        handler_name="on_property_inspector_did_appear",
-    ),
-    "propertyInspectorDidDisappear": EventRoutingObj(
-        obj=events_received_objs.PropertyInspectorDidDisappear,
-        handler_name="on_property_inspector_did_disappear",
-    ),
-    "sendToPlugin": EventRoutingObj(
-        obj=events_received_objs.SendToPlugin,
-        handler_name="on_send_to_plugin",
-    ),
-    "sendToPropertyInspector": EventRoutingObj(
-        obj=events_received_objs.SendToPropertyInspector,
-        handler_name="on_send_to_property_inspector",
-    ),
-}
 
-PLUGIN_EVENT_ROUTING_MAP: Dict[str, EventRoutingObj] = {
-    "didReceiveGlobalSettings": EventRoutingObj(
-        obj=events_received_objs.DidReceiveGlobalSettings,
-        handler_name="on_did_receive_global_settings",
-    ),
-    "deviceDidConnect": EventRoutingObj(
-        obj=events_received_objs.DeviceDidConnect,
-        handler_name="on_device_did_connect",
-    ),
-    "deviceDidDisconnect": EventRoutingObj(
-        obj=events_received_objs.DeviceDidDisconnect,
-        handler_name="on_device_did_disconnect",
-    ),
-    "applicationDidLaunch": EventRoutingObj(
-        obj=events_received_objs.ApplicationDidLaunch,
-        handler_name="on_application_did_launch",
-    ),
-    "applicationDidTerminate": EventRoutingObj(
-        obj=events_received_objs.ApplicationDidTerminate,
-        handler_name="on_application_did_terminate",
-    ),
-    "systemDidWakeUp": EventRoutingObj(
-        obj=events_received_objs.SystemDidWakeUp,
-        handler_name="on_system_did_wake_up",
-    ),
-}
-
-EVENT_ROUTING_MAP: Dict[str, EventRoutingObj] = {
-    **ACTION_EVENT_ROUTING_MAP,
-    **PLUGIN_EVENT_ROUTING_MAP,
-}
+class EventRoutingObj(BaseModel):
+    handler_name: str
+    obj_type: Type[BaseModel]
+    type: EventRoutingObjTypes
+
+
+EVENT_ROUTING_MAP: Dict[str, EventRoutingObj] = {}
+
+
+def fill_routing_map(
+        routing_map: Dict[str, EventRoutingObj],
+        event_handler_mixin: mixins.BaseEventHandlerMixin,
+        event_routing_obj_type: EventRoutingObjTypes,
+) -> None:
+    event_handler_mixin_dict = dict(event_handler_mixin.__dict__)
+    for attr, value in event_handler_mixin_dict.items():
+        attr: str
+        if not attr.startswith("on_"):
+            continue
+        assert callable(value)
+
+        handler_type_hints = get_type_hints(value)
+        obj_type = handler_type_hints["obj"]
+        obj = obj_type.construct()
+        event_name = obj.event
+        handler_name = value.__name__
+        routing_map[event_name] = EventRoutingObj(
+            handler_name=handler_name,
+            obj_type=obj_type,
+            type=event_routing_obj_type,
+        )
+
+
+def fill_event_routing_map() -> None:
+    for event_handler_mixin, event_routing_obj_type in (
+            (mixins.ActionEventHandlersMixin, EventRoutingObjTypes.ACTION),
+            (mixins.PluginEventHandlersMixin, EventRoutingObjTypes.PLUGIN),
+    ):
+        fill_routing_map(
+            routing_map=EVENT_ROUTING_MAP,
+            event_handler_mixin=event_handler_mixin,
+            event_routing_obj_type=event_routing_obj_type,
+        )
+
+
+fill_event_routing_map()

src/streamdeck_sdk/logger.py

@@ -5,16 +5,18 @@
 
 from decohints import decohints
 
-logger: logging.Logger = logging.getLogger('default')
+_root_logger: logging.Logger = logging.getLogger()
+_log_errors_decorator_logger = logging.getLogger("log_errors_decorator")
+logger = logging.getLogger("streamdeck_plugin")
 
 
-def init_logger(
+def init_root_logger(
         log_file: Path,
         log_level: int = logging.DEBUG,
         log_max_bytes: int = 3 * 1024 * 1024,
         log_backup_count: int = 2,
 ) -> None:
-    logger.setLevel(log_level)
+    _root_logger.setLevel(log_level)
     logs_dir: Path = log_file.parent
     logs_dir.mkdir(parents=True, exist_ok=True)
     rfh = RotatingFileHandler(
@@ -30,7 +32,7 @@ def init_logger(
         "%(asctime)s - [%(levelname)s] - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d): %(message)s"
     )
     rfh.setFormatter(formatter)
-    logger.addHandler(rfh)
+    _root_logger.addHandler(rfh)
 
 
 @decohints
@@ -44,8 +46,12 @@ def wrapper(*args, **kwargs):
         try:
             result = func(*args, **kwargs)
         except BaseException as err:
-            logger.error(str(err), exc_info=True)
+            _log_errors_decorator_logger.error(str(err), exc_info=True)
             return
         return result
 
     return wrapper
+
+
+def rename_plugin_logger(name: str):
+    logger.name = name

src/streamdeck_sdk/mixins.py

@@ -5,8 +5,7 @@
 import websocket
 from websocket import ABNF
 
-from . import events_received_objs
-from . import events_sent_objs
+from .sd_objs import events_received_objs, events_sent_objs
 from .logger import log_errors
 
 
@@ -26,7 +25,11 @@ def send(
         self.ws.send(data, opcode)
 
 
-class PluginEventsSentMixin(SendMixin):
+class BaseEventSendMixin(SendMixin):
+    pass
+
+
+class PluginEventsSendMixin(BaseEventSendMixin):
     plugin_uuid: str
 
     def set_global_settings(self, payload: dict) -> None:
@@ -73,7 +76,7 @@ def switch_to_profile(
         self.send(message)
 
 
-class ActionEventsSentMixin(SendMixin):
+class ActionEventsSendMixin(BaseEventSendMixin):
     def set_settings(
             self,
             context: str,
@@ -185,7 +188,11 @@ def send_to_property_inspector(
         self.send(message)
 
 
-class ActionEventHandlersMixin:
+class BaseEventHandlerMixin:
+    pass
+
+
+class ActionEventHandlersMixin(BaseEventHandlerMixin):
     def on_did_receive_settings(self, obj: events_received_objs.DidReceiveSettings) -> None:
         pass
 
@@ -232,7 +239,7 @@ def on_send_to_property_inspector(self, obj: events_received_objs.SendToProperty
         pass
 
 
-class PluginEventHandlersMixin:
+class PluginEventHandlersMixin(BaseEventHandlerMixin):
     def on_did_receive_global_settings(self, obj: events_received_objs.DidReceiveGlobalSettings) -> None:
         pass
 

src/streamdeck_sdk/sd_objs/__init__.py

@@ -0,0 +1,3 @@
+from . import events_received_objs
+from . import events_sent_objs
+from . import registration_objs