Change to use asyncio rather than threading

This allows event-driven things to be far more easily driven
(ie. pulsectl)

Also, Rather than defining a source / sync for the mute buttons,
use the current default one - how often do you need to immediately mute
a device that's not in use?

Author: whi-tw

.gitignore

@@ -4,3 +4,4 @@
 __pycache__/
 htmlcov/
 venv/
+.vscode

.python-version

@@ -0,0 +1 @@
+3.10.0

devdeck/controls/clock_control.py

@@ -1,24 +1,23 @@
+import asyncio
+from asyncio.events import get_event_loop
 import threading
 from datetime import datetime
-from time import sleep
+from asyncio import sleep
 
 from devdeck_core.controls.deck_control import DeckControl
 
 
 class ClockControl(DeckControl):
 
-    def __init__(self, key_no, **kwargs):
+    def __init__(self, key_no: int, **kwargs):
+        self.loop = get_event_loop()
         super().__init__(key_no, **kwargs)
-        self.thread = None
-        self.running = False
 
     def initialize(self):
-        self.thread = threading.Thread(target=self._update_display)
-        self.running = True
-        self.thread.start()
+        self.loop.create_task(self._update_display())
 
-    def _update_display(self):
-        while self.running is True:
+    async def _update_display(self):
+        while True:
             with self.deck_context() as context:
                 now = datetime.now()
 
@@ -33,10 +32,4 @@ def _update_display(self):
                         .center_vertically(100) \
                         .font_size(75) \
                         .end()
-            sleep(1)
-
-    def dispose(self):
-        self.running = False
-        if self.thread:
-            self.thread.join()
-
+            await sleep(1)

devdeck/controls/command_control.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 from subprocess import Popen, DEVNULL
@@ -19,4 +20,4 @@ def pressed(self):
         try:
             Popen(self.settings['command'], stdout=DEVNULL, stderr=DEVNULL)
         except Exception as ex:
-            self.__logger.error("Error executing command %s: %s", self.settings['command'], str(ex))
+            self.__logger.error("Error executing command %s: %s", self.settings['command'], str(ex))

devdeck/controls/mic_mute_control.py

@@ -1,66 +1,70 @@
+from asyncio import sleep
+from asyncio.events import get_event_loop
 import logging
 import os
 
-from pulsectl import pulsectl
+import pulsectl_asyncio
 
 from devdeck_core.controls.deck_control import DeckControl
 
 
 class MicMuteControl(DeckControl):
 
     def __init__(self, key_no, **kwargs):
+        self.loop = get_event_loop()
         self.pulse = None
         self.__logger = logging.getLogger('devdeck')
         super().__init__(key_no, **kwargs)
 
-    def initialize(self):
+    async def _init(self):
         if self.pulse is None:
-            self.pulse = pulsectl.Pulse('MicMuteControl')
-        self.__render_icon()
+            self.pulse = pulsectl_asyncio.PulseAsync('MicMuteControl')
+            await self.pulse.connect()
+        self.loop.create_task(self._update_display())
+
+    def initialize(self):
+        self.loop.create_task(self._init())
 
     def pressed(self):
-        mic = self.__get_mic()
-        if mic is None:
-            return
-        self.pulse.source_mute(mic.index, mute=(not mic.mute))
-        self.__render_icon()
+        self.loop.create_task(self._handle_mute())
 
-    def __get_mic(self):
-        sources = self.pulse.source_list()
+    async def _handle_mute(self):
+        mic = await self._get_source()
+        self.loop.create_task(self.pulse.source_mute(mic.index, mute=(not mic.mute)))
 
-        selected_mic = [mic for mic in sources if mic.description == self.settings['microphone']]
-        if len(selected_mic) == 0:
-            possible_mics = [output.description for output in sources]
-            self.__logger.warning("Microphone '%s' not found in list of possible inputs:\n%s",
-                                  self.settings['microphone'],
-                                  '\n'.join(possible_mics))
-            return None
-        return selected_mic[0]
+    async def _get_source(self):
+        sources = await self.pulse.source_list()
+        server_info = await self.pulse.server_info()
+        default_source_name = server_info.default_source_name
+        return next((source for source in sources if source.name == default_source_name), None)
 
-    def __render_icon(self):
+    async def _update_display(self):
         with self.deck_context() as context:
-            mic = self.__get_mic()
-            if mic is None:
-                with context.renderer() as r:
-                    r \
-                        .text('MIC \nNOT FOUND') \
-                        .color('red') \
-                        .center_vertically() \
-                        .center_horizontally() \
-                        .font_size(85) \
-                        .text_align('center') \
-                        .end()
-                return
-            if mic.mute == 0:
-                with context.renderer() as r:
-                    r.image(os.path.join(os.path.dirname(__file__), "../assets/font-awesome", 'microphone.png')).end()
-            else:
+            while True:
+                mic = await self._get_source()
                 with context.renderer() as r:
-                    r.image(os.path.join(os.path.dirname(__file__), "../assets/font-awesome", 'microphone-mute.png')).end()
+                    try:
+                        match mic.mute:
+                            case 0:
+                                r.image(os.path.join(os.path.dirname(__file__),
+                                                    "../assets/font-awesome", 'microphone.png')).end()
+                            case 1:
+                                r.image(os.path.join(os.path.dirname(__file__),
+                                        "../assets/font-awesome", 'microphone-mute.png')).end()
+                    except AttributeError:
+                        r \
+                            .text('MIC \nNOT FOUND') \
+                            .color('red') \
+                            .center_vertically() \
+                            .center_horizontally() \
+                            .font_size(85) \
+                            .text_align('center') \
+                            .end()
+                await sleep(0.1)
 
     def settings_schema(self):
         return {
             'microphone': {
                 'type': 'string'
             }
-        }
+        }

devdeck/controls/name_list_control.py

@@ -1,8 +1,8 @@
+import asyncio
 import os
 
 from devdeck_core.controls.deck_control import DeckControl
 
-
 class NameListControl(DeckControl):
 
     def __init__(self, key_no, **kwargs):

devdeck/controls/timer_control.py

@@ -1,62 +1,80 @@
-import datetime
+from asyncio.events import get_event_loop
+from datetime import datetime
 import os
-import threading
+import enum
+import asyncio
 from time import sleep
 
 from devdeck_core.controls.deck_control import DeckControl
 
 
+class TimerState(enum.Enum):
+    RUNNING = 1
+    STOPPED = 2
+    RESET = 3
+
+
 class TimerControl(DeckControl):
 
     def __init__(self, key_no, **kwargs):
-        self.start_time = None
-        self.end_time = None
-        self.thread = None
-        super().__init__(key_no, **kwargs)
+        self.loop = get_event_loop()
+        self.start_time: datetime = None
+        self.end_time: datetime = None
+        self.state = TimerState.RESET
+        super().__init__(key_no, ** kwargs)
 
     def initialize(self):
+        self.loop.create_task(self._update_display())
         with self.deck_context() as context:
             with context.renderer() as r:
-                r.image(os.path.join(os.path.dirname(__file__), "../assets/font-awesome", 'stopwatch.png')).end()
+                r.image(os.path.join(os.path.dirname(__file__),
+                        "../assets/font-awesome", 'stopwatch.png')).end()
 
     def pressed(self):
-        if self.start_time is None:
-            self.start_time = datetime.datetime.now()
-            self.thread = threading.Thread(target=self._update_display)
-            self.thread.start()
-        elif self.end_time is None:
-            self.end_time = datetime.datetime.now()
-            self.thread.join()
-            with self.deck_context() as context:
-                with context.renderer() as r:
-                    r.text(TimerControl.time_diff_to_str(self.end_time - self.start_time))\
-                        .font_size(120)\
-                        .color('red')\
-                        .center_vertically().center_horizontally().end()
-        else:
-            self.start_time = None
-            self.end_time = None
-            with self.deck_context() as context:
-                with context.renderer() as r:
-                    r.image(os.path.join(
-                        os.path.join(os.path.dirname(__file__), "../assets/font-awesome", 'stopwatch.png'))).end()
-
-    def _update_display(self):
-        while self.end_time is None:
-            if self.start_time is None:
-                sleep(1)
-                continue
-            cutoff = datetime.datetime.now() if self.end_time is None else self.end_time
+        match self.state:
+            case TimerState.RESET:
+                self.start_time = datetime.now()
+                self.end_time = None
+                self.state = TimerState.RUNNING
+            case TimerState.RUNNING:
+                if not self.start_time:
+                    raise Exception("how did you get here?")
+                self.end_time = datetime.now()
+                self.state = TimerState.STOPPED
+            case TimerState.STOPPED:
+                self.start_time = self.end_time = None
+                self.state = TimerState.RESET
+
+    async def _update_display(self, repeat=True):
+        while True:
             with self.deck_context() as context:
                 with context.renderer() as r:
-                    r.text(TimerControl.time_diff_to_str(cutoff - self.start_time)) \
-                        .font_size(120) \
-                        .center_vertically().center_horizontally().end()
-            sleep(1)
+                    match self.state:
+                        case TimerState.RUNNING:
+                            r.text(TimerControl.time_diff_to_str(datetime.now() - self.start_time))\
+                                .font_size(120)\
+                                .color('red')\
+                                .center_vertically().center_horizontally().end()
+                        case TimerState.STOPPED:
+                            r.text(TimerControl.time_diff_to_str(self.end_time - self.start_time))\
+                                .font_size(120)\
+                                .color('yellow')\
+                                .center_vertically().center_horizontally().end()
+                        case _:
+                            r.image(os.path.join(
+                                os.path.join(os.path.dirname(__file__), "../assets/font-awesome", 'stopwatch.png'))).end()
+            if repeat:
+                await asyncio.sleep(0.1)
+            else:
+                return
 
     @staticmethod
     def time_diff_to_str(diff):
-        seconds = diff.total_seconds()
-        minutes, seconds = divmod(seconds, 60)
+        total_seconds = diff.total_seconds()
+        minutes, seconds = divmod(total_seconds, 60)
         hours, minutes = divmod(minutes, 60)
+        if total_seconds < 60:
+            return f'{int(seconds):02d}'
+        elif total_seconds < 3600:
+            return f'{int(minutes):02d}:{int(seconds):02d}'
         return f'{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}'

devdeck/controls/volume_level_control.py

@@ -1,3 +1,5 @@
+import asyncio
+from asyncio.events import get_event_loop
 import logging
 import os
 
@@ -8,11 +10,12 @@
 
 class VolumeLevelControl(DeckControl):
 
-    def __init__(self, key_no, **kwargs):
+    def __init__(self, key_no,  **kwargs):
+        self.loop = get_event_loop(),
         self.pulse = None
         self.volume = None
         self.__logger = logging.getLogger('devdeck')
-        super().__init__(key_no, **kwargs)
+        super().__init__(key_no, ** kwargs)
 
     def initialize(self):
         if self.pulse is None:
@@ -29,10 +32,12 @@ def pressed(self):
 
     def __get_output(self):
         sinks = self.pulse.sink_list()
-        selected_output = [output for output in sinks if output.description == self.settings['output']]
+        selected_output = [
+            output for output in sinks if output.description == self.settings['output']]
         if len(selected_output) == 0:
             possible_ouputs = [output.description for output in sinks]
-            self.__logger.warning("Output '%s' not found in list of possible outputs:\n%s", self.settings['output'], '\n'.join(possible_ouputs))
+            self.__logger.warning("Output '%s' not found in list of possible outputs:\n%s",
+                                  self.settings['output'], '\n'.join(possible_ouputs))
             return None
         return selected_output[0]
 
@@ -72,4 +77,4 @@ def settings_schema(self):
             'volume': {
                 'type': 'integer'
             }
-        }
+        }