From 32ab171086758cbbd6d25d37c1200f1209129ea9 Mon Sep 17 00:00:00 2001 From: Kalle-Wirsch Date: Sun, 24 Feb 2019 18:23:10 +0100 Subject: [PATCH 1/4] Reintegrated changes from last Pull request --- src/StreamDeck/DeviceManager.py | 2 +- src/StreamDeck/Devices/StreamDeck.py | 18 +- src/StreamDeck/Devices/StreamDeckFilter.py | 188 +++++++++++++++++++++ src/StreamDeck/StreamDeckFilter.py | 188 +++++++++++++++++++++ src/example.py | 14 +- 5 files changed, 398 insertions(+), 12 deletions(-) create mode 100644 src/StreamDeck/Devices/StreamDeckFilter.py create mode 100644 src/StreamDeck/StreamDeckFilter.py diff --git a/src/StreamDeck/DeviceManager.py b/src/StreamDeck/DeviceManager.py index de6468f..4f65d4f 100644 --- a/src/StreamDeck/DeviceManager.py +++ b/src/StreamDeck/DeviceManager.py @@ -9,7 +9,7 @@ from .Devices.StreamDeckMini import StreamDeckMini from .Transport.Dummy import Dummy from .Transport.HIDAPI import HIDAPI - +from .StreamDeckFilter import TempoFilter, StateChangedFilter, DebounceFilter class DeviceManager: """ diff --git a/src/StreamDeck/Devices/StreamDeck.py b/src/StreamDeck/Devices/StreamDeck.py index b9c2e2e..d5797fc 100644 --- a/src/StreamDeck/Devices/StreamDeck.py +++ b/src/StreamDeck/Devices/StreamDeck.py @@ -6,9 +6,8 @@ # from abc import ABC, abstractmethod - import threading - +from .StreamDeckFilter import TempoFilter, StateChangedFilter, DebounceFilter class StreamDeck(ABC): """ @@ -30,10 +29,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 +67,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): """ diff --git a/src/StreamDeck/Devices/StreamDeckFilter.py b/src/StreamDeck/Devices/StreamDeckFilter.py new file mode 100644 index 0000000..07a9483 --- /dev/null +++ b/src/StreamDeck/Devices/StreamDeckFilter.py @@ -0,0 +1,188 @@ +# 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 diff --git a/src/StreamDeck/StreamDeckFilter.py b/src/StreamDeck/StreamDeckFilter.py new file mode 100644 index 0000000..07a9483 --- /dev/null +++ b/src/StreamDeck/StreamDeckFilter.py @@ -0,0 +1,188 @@ +# 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 diff --git a/src/example.py b/src/example.py index cb799fe..8f2eae0 100755 --- a/src/example.py +++ b/src/example.py @@ -8,11 +8,19 @@ # 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): @@ -27,7 +35,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 +49,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} From 3339229bd581718f8bd708252e355f198767aaec Mon Sep 17 00:00:00 2001 From: Kalle-Wirsch Date: Sun, 24 Feb 2019 18:51:13 +0100 Subject: [PATCH 2/4] Duplicate file --- src/StreamDeck/StreamDeckFilter.py | 188 ----------------------------- 1 file changed, 188 deletions(-) delete mode 100644 src/StreamDeck/StreamDeckFilter.py diff --git a/src/StreamDeck/StreamDeckFilter.py b/src/StreamDeck/StreamDeckFilter.py deleted file mode 100644 index 07a9483..0000000 --- a/src/StreamDeck/StreamDeckFilter.py +++ /dev/null @@ -1,188 +0,0 @@ -# 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 From d537d132c734a6f794514df2a981fbba93b458d8 Mon Sep 17 00:00:00 2001 From: Kalle-Wirsch Date: Sun, 24 Feb 2019 18:57:52 +0100 Subject: [PATCH 3/4] Unnecessary import --- src/StreamDeck/DeviceManager.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/StreamDeck/DeviceManager.py b/src/StreamDeck/DeviceManager.py index 4f65d4f..98449ef 100644 --- a/src/StreamDeck/DeviceManager.py +++ b/src/StreamDeck/DeviceManager.py @@ -9,7 +9,6 @@ from .Devices.StreamDeckMini import StreamDeckMini from .Transport.Dummy import Dummy from .Transport.HIDAPI import HIDAPI -from .StreamDeckFilter import TempoFilter, StateChangedFilter, DebounceFilter class DeviceManager: """ From c7bafe920e6c0bbac93e6fedc87d86e08af470bf Mon Sep 17 00:00:00 2001 From: Kalle-Wirsch Date: Sun, 24 Feb 2019 20:11:20 +0100 Subject: [PATCH 4/4] Flake8 --- src/StreamDeck/DeviceManager.py | 1 + src/StreamDeck/Devices/StreamDeck.py | 3 +- src/StreamDeck/Devices/StreamDeckFilter.py | 42 +++++++++++----------- src/example.py | 4 ++- 4 files changed, 28 insertions(+), 22 deletions(-) diff --git a/src/StreamDeck/DeviceManager.py b/src/StreamDeck/DeviceManager.py index 98449ef..de6468f 100644 --- a/src/StreamDeck/DeviceManager.py +++ b/src/StreamDeck/DeviceManager.py @@ -10,6 +10,7 @@ from .Transport.Dummy import Dummy from .Transport.HIDAPI import HIDAPI + class DeviceManager: """ Central device manager, to enumerate any attached StreamDeck devices. An diff --git a/src/StreamDeck/Devices/StreamDeck.py b/src/StreamDeck/Devices/StreamDeck.py index d5797fc..ae55370 100644 --- a/src/StreamDeck/Devices/StreamDeck.py +++ b/src/StreamDeck/Devices/StreamDeck.py @@ -7,7 +7,8 @@ from abc import ABC, abstractmethod import threading -from .StreamDeckFilter import TempoFilter, StateChangedFilter, DebounceFilter +from .StreamDeckFilter import StateChangedFilter + class StreamDeck(ABC): """ diff --git a/src/StreamDeck/Devices/StreamDeckFilter.py b/src/StreamDeck/Devices/StreamDeckFilter.py index 07a9483..aa39c1e 100644 --- a/src/StreamDeck/Devices/StreamDeckFilter.py +++ b/src/StreamDeck/Devices/StreamDeckFilter.py @@ -10,7 +10,7 @@ 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, +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) @@ -43,7 +43,7 @@ You can simply subclass CallbackFilter (or any other of the above filter classes) and override CallBackFilter.map_states(self, new_state) -JSON support: +JSON support: See JSONexample.py Known Bugs: @@ -64,6 +64,7 @@ from time import time from functools import wraps + def manage_states(f): """ decorator function for CallbackFilter.mapstates() @@ -86,7 +87,7 @@ def updater(self, new_state, *args, **kwargs): 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() """ @@ -97,7 +98,7 @@ def __init__(self, default_state=False): self._last_key_time = time() # time of last callback self._delay = 0.0 # delay since last callback - @manage_states # register states, time and delay + @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 @@ -109,9 +110,9 @@ def map_states(self, new_state): 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 + Designed to be part of json.Encoder.default(self, obj): -> add the following lines ... if isinstance(obj, CallBackFilter): return obj.json_serialize() @@ -121,22 +122,23 @@ def json_serialize(self): @classmethod def json_deserialize(cls, data_dict): - """ - Creates obj and updates state. + """ + 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): + if isinstance(x, dict): key, value = next(iter(x.items())) cbf_class = globals()[key] - return cbf_class.json_deserialize(value) + return cbf_class.json_deserialize(value) ... """ - obj = cls() # create object + 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 """ @@ -147,8 +149,8 @@ def map_states(self, new_state): class DebounceFilter(CallBackFilter): - """ - DebounceFilter: Returns None (i.e. no callback) if stade did not change + """ + 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. """ @@ -163,14 +165,14 @@ def map_states(self, new_state): 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. + like W,A,S,D, etc. """ def __init__(self, default_state=False, key_delay=0.003, tempo_delay=0.3): @@ -179,10 +181,10 @@ def __init__(self, default_state=False, key_delay=0.003, tempo_delay=0.3): @manage_states def map_states(self, new_state): - rv=None + 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_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 + 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 diff --git a/src/example.py b/src/example.py index 8f2eae0..ff221e8 100755 --- a/src/example.py +++ b/src/example.py @@ -23,6 +23,8 @@ def ospath_to_module(): # 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) @@ -35,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(os.path.join(ospath_to_module(),"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))