Add various Filters for Key Callback and and provide a Framework to create new Filters

src/StreamDeck/Devices/StreamDeck.py

@@ -6,8 +6,8 @@
 #
 
 from abc import ABC, abstractmethod
-
 import threading
+from .StreamDeckFilter import StateChangedFilter
 
 
 class StreamDeck(ABC):
@@ -30,10 +30,14 @@ class StreamDeck(ABC):
 
     def __init__(self, device):
         self.device = device
-        self.last_key_states = [False] * self.KEY_COUNT
+        self.filter = [StateChangedFilter() for _ in range(self.KEY_COUNT)]
         self.read_thread = None
         self.key_callback = None
 
+    @property  # provide property last_key_states for backwards compatibility
+    def last_key_states(self):
+        return [self.filter[k]._last_cb_state for k in range(self.KEY_COUNT)]
+
     def __del__(self):
         """
         Deletion handler for the StreamDeck, automatically closing the transport
@@ -64,11 +68,10 @@ def _read(self):
                 new_key_states = [bool(s) for s in payload[1:]]
 
                 if self.key_callback is not None:
-                    for k, (old, new) in enumerate(zip(self.last_key_states, new_key_states)):
-                        if old != new:
-                            self.key_callback(self, k, new)
-
-                self.last_key_states = new_key_states
+                    for k, (new, cbfilter) in enumerate(zip(new_key_states, self.filter)):
+                        new_state = cbfilter.map_states(new)
+                        if new_state is not None:
+                            self.key_callback(self, k, new_state)
 
     def _setup_reader(self, callback):
         """

src/StreamDeck/Devices/StreamDeckFilter.py

@@ -0,0 +1,190 @@
+#  Python Stream Deck Library - FilterPlugins
+#      Released under the MIT license
+#
+#      https://github.com/Kalle-Wirsch
+#
+
+"""
+Module StreamDeckFilter
+
+StreamDeckFilter is designed to modify the callback behavior of the StreamDeck class while
+hiding the internal changes to existing implementations.
+
+Generally StreamDeck invokes the Callback-Function, whenever a key-status changed,
+i.e. a key is pressed or released. The callback can be set with deck.set_key_callback(self, callback),
+and looks like this:
+    key_callback(self, k, new_state)
+        where k is the changed key and new_state is the new key state
+        (pressed = True, released = False)
+
+Usage:
+    StreamDeckFilter allows to change the callback behavior by assigning a specific Filter to each key as follows:
+        deck.filter[key] = TempoFilter()
+
+Implemented filters:
+    CallBackFilter()
+        Base class, that calls the callback function on each internal device read
+
+    StateChangedFilter()
+        The callback function is called only if a key k changed its state
+
+    DebounceFilter(default_state=False, key_delay=0.003)
+        the callbackfunction is called only,
+        if the state did not already change within the last key_delay seconds
+        Therefore filtering out fast state chenges do to "key chattering",
+        i.e. fast state changes, when the key is "almost" pressed.
+
+    TempoFilter(default_state=False, key_delay=0.003, tempo_delay=0.3)
+        As Debouncefilter but returns
+        - True if the key is pressed less then tempo_delay seconds
+        - False if key is pressed for more than tempo-delay seconds
+
+Extendability:
+    You can simply subclass CallbackFilter (or any other of the above filter classes)
+    and override CallBackFilter.map_states(self, new_state)
+
+JSON support:
+    See JSONexample.py
+
+Known Bugs:
+    Debounce filter doesn't work well with long key_delays.
+    It should reset to False after key_delay seconds or at least fter some internal delay
+    or clock of of an MVC View-Refresh
+
+Future Development plans
+    I will add more filters in the future e.g.
+
+    HoldFilter
+        changes the state if key is kept pressed/released for countdown_delay seconds
+    CountDownFilter
+        changes the state after countdown_delay time, but calls back every count_delay seconds
+
+"""
+
+from time import time
+from functools import wraps
+
+
+def manage_states(f):
+    """
+    decorator function for CallbackFilter.mapstates()
+    registers state, time and delay of last callback
+    """
+    @wraps(f)       # make updater look like decorated function in traceback
+    def updater(self, new_state, *args, **kwargs):
+        # registere delay inside CallbackFilter.mapstates
+        self._delay = time() - self._last_key_time
+        rv = f(self, new_state, *args, **kwargs)   # call function
+        if rv is not None:
+            # register old physical state inside CallbackFilter.mapstates
+            self._last_key_state = new_state
+            # register old callback state inside CallbackFilter.mapstates
+            self._last_cb_state = rv
+            # register time inside CallbackFilter.mapstates
+            self._last_key_time = time()
+        return rv
+    return updater
+
+
+class CallBackFilter(object):
+    """
+    Serves as base for Filters that map physical states (True, False) to a different range, type of values
+    Simply assign a filter to a specific key like this:  deck.filter[key] = StateChangedFilter()
+    """
+
+    def __init__(self, default_state=False):
+        self._last_key_state = False            # last physical state callback time
+        self._last_cb_state = default_state     # last state reported to callback function
+        self._last_key_time = time()            # time of last callback
+        self._delay = 0.0                       # delay since last callback
+
+    @manage_states  # register states, time and delay
+    def map_states(self, new_state):
+        """
+        - maps/filers pysical states(pressed=True, released=False) to before callback to client
+        - stores old states, times and delays(since last callback)
+        - in principle can be overridden to return anything e.g.
+        Enum('MyStates', 'LONG_PRESS, SHORT_PRESSED, NOTPRESSED_FOR_300_MILLISECONDS, PRESSED_5_TIMES_IN_5_MINUTES, ...')
+        Assumption: map_states returns None <= > no callback will be made
+        """
+        return new_state  # do not filter anything
+
+    def json_serialize(self):
+        """
+        Serialize only non-protected, i.e. non volatile attributes.
+        Designed to be part of json.Encoder.default(self, obj): -> add the following lines
+            ...
+            if isinstance(obj, CallBackFilter):
+                return obj.json_serialize()
+            ...
+        """
+        return {self.__class__.__name__: {k: v for k, v in self.__dict__.items() if k[:1] != '_'}}
+
+    @classmethod
+    def json_deserialize(cls, data_dict):
+        """
+        Creates obj and updates state.
+        Add the following lines to decode a loaded object instance from native json.load to CallBackFilter
+            ...
+            if isinstance(x, dict):
+                key, value = next(iter(x.items()))
+                cbf_class = globals()[key]
+                return cbf_class.json_deserialize(value)
+            ...
+        """
+        obj = cls()  # create object
+        obj.__dict__.update(data_dict)  # update native json objects
+        return obj
+
+
+class StateChangedFilter(CallBackFilter):
+    """
+    StateChangeFilter: Returns None (i.e. no callback) if stade did not change
+    equals the default StreamDeck behavior
+    """
+    @manage_states  # register states, time and delay
+    def map_states(self, new_state):
+        # only return on state change
+        return new_state if self._last_key_state != new_state else None
+
+
+class DebounceFilter(CallBackFilter):
+    """
+    DebounceFilter: Returns None (i.e. no callback) if stade did not change
+    or did already change within the last self.key_delay seconds
+    - prevents chattering of keys.
+    """
+    def __init__(self, default_state=False, key_delay=0.003):
+        self.key_delay = key_delay
+        CallBackFilter.__init__(self, default_state)
+
+    @manage_states
+    def map_states(self, new_state):
+        # only return on state change and if delay since last callbak >= key delay
+        return new_state if self._last_key_state != new_state and self._delay >= self.key_delay else None
+
+
+class TempoFilter(DebounceFilter):
+    """
+    TempoFilter: returns
+    - True if key is pressed for less than tempo_delay seconds
+    - False if key is pressed for more than tempo_delay seconds
+    - None if key pressed for less than key_delay seconds or state did not change
+    Note: In tempo mode callbacks are fired at key release time
+        i.e. it results in lag/latency and is not suitable gaming keys
+        like W,A,S,D, etc.
+    """
+
+    def __init__(self, default_state=False, key_delay=0.003, tempo_delay=0.3):
+        self.tempo_delay = tempo_delay
+        DebounceFilter.__init__(self, default_state)
+
+    @manage_states
+    def map_states(self, new_state):
+        rv = None
+        if self._last_key_state != new_state and self._delay >= self.key_delay:
+            self._last_key_time = time()  # we always need to register last state change and time
+            self._last_key_state = new_state
+            if new_state is False:        # but only return a callback on key release
+                rv = True if self._delay < self.tempo_delay else False
+        return rv

src/example.py

@@ -8,13 +8,23 @@
 #
 
 import threading
+import os
+from inspect import getsourcefile
 from StreamDeck.DeviceManager import DeviceManager
 from StreamDeck.ImageHelpers import PILHelper
 from PIL import Image, ImageDraw, ImageFont
 
 
+# returns absolute path to this module
+# allows to load files relative to module instead of relative to current working directory
+# i.e. module does not need to be placed in current working directory
+def ospath_to_module():
+    return os.path.dirname(getsourcefile(lambda: 0))
+
 # Generates a custom tile with run-time generated text and custom image via the
 # PIL module.
+
+
 def render_key_image(deck, icon_filename, label_text):
     # Create new key image of the correct dimensions, black background
     image = PILHelper.create_image(deck)
@@ -27,7 +37,7 @@ def render_key_image(deck, icon_filename, label_text):
 
     # Load a custom TrueType font and use it to overlay the key index, draw key
     # number onto the image
-    font = ImageFont.truetype("Assets/Roboto-Regular.ttf", 14)
+    font = ImageFont.truetype(os.path.join(ospath_to_module(), "Assets", "Roboto-Regular.ttf"), 14)
     draw = ImageDraw.Draw(image)
     draw.text((10, image.height - 20), text=label_text, font=font, fill=(255, 255, 255, 128))
 
@@ -41,11 +51,11 @@ def get_key_style(deck, key, state):
 
     if key == exit_key_index:
         name = "exit"
-        icon = "Assets/{}.png".format("Exit")
+        icon = os.path.join(ospath_to_module(), "Assets", "{}.png".format("Exit"))
         text = "Bye" if state else "Exit"
     else:
         name = "emoji"
-        icon = "Assets/{}.png".format("Pressed" if state else "Released")
+        icon = os.path.join(ospath_to_module(), "Assets", "{}.png".format("Pressed" if state else "Released"))
         text = "Pressed!" if state else "Key {}".format(key)
 
     return {"name": name, "icon": icon, "label": text}