diff --git a/client.py b/client.py index 0334597..2e73f3b 100644 --- a/client.py +++ b/client.py @@ -7,36 +7,14 @@ import time import threading import signal import argparse - import subprocess - import RNS import LXMF + +from PIL import Image, ImageDraw, ImageFont from LXST.Primitives.Telephony import Telephone - -# ── Whisplay HAT GUI support ────────────────────────────────────────── -_RUNTIME_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "runtime") -if os.path.isdir(_RUNTIME_DIR): - sys.path.insert(0, _RUNTIME_DIR) - -_WHISPLAY_AVAILABLE = False -try: - from whisplay_client import create_whisplay_hardware - _WHISPLAY_AVAILABLE = True -except ImportError: - pass - -_HAS_PIL = False -if _WHISPLAY_AVAILABLE: - try: - from PIL import Image, ImageDraw, ImageFont - _HAS_PIL = True - except ImportError: - pass - - - +from whisplay_interface import WhisplayInterface class Terminal: if not RNS.vendor.platformutils.is_windows(): @@ -67,19 +45,15 @@ class ChannelClient: self._response_event = threading.Event() # ── Whisplay HAT init ── - self.board = None self.selected_index = 0 self._wm8960_device = None self._ptt_enabled = True self._ptt_active = False - self.gestures = None + self.whisplay_interface = None + self._gesture_thread = None - if use_whisplay and _WHISPLAY_AVAILABLE and _HAS_PIL: - try: - self._init_whisplay() - except Exception as e: - RNS.log(f"Whisplay init failed: {e}", RNS.LOG_WARNING) - self.board = None + if use_whisplay: + self._init_whisplay() if configdir is None: if os.path.isdir("/etc/lxst-client"): @@ -143,25 +117,16 @@ class ChannelClient: # ── Whisplay HAT GUI ───────────────────────────────────────────── def _init_whisplay(self): - from whisplay_client import create_whisplay_hardware - - self.board = create_whisplay_hardware( + self.whisplay_interface = WhisplayInterface() + self.whisplay_interface.create_board( app_id="lxst-client", display_name="LXST Client", icon="LC", use_daemon_default_log=True, ) - self.board.set_backlight(70) - from whisplay_interface import WhisplayInterface - self.gestures = WhisplayInterface(self.board) - self.gestures.on_press(self._on_button_press) - self.gestures.on_release(self._on_button_release) - if hasattr(self.board, "on_exit_request"): - self.board.on_exit_request(self._on_exit_request) - if hasattr(self.board, "on_focus_revoked"): - self.board.on_focus_revoked(self._on_focus_revoked) - if hasattr(self.board, "start_event_listener"): - self.board.start_event_listener() + self.whisplay_interface.board.set_backlight(70) + # self.whisplay_interface.board.on_exit_request(self._on_exit_request) + # self.whisplay_interface.board.on_focus_revoked(self._on_focus_revoked) self._load_fonts() self._setup_whisplay_audio() @@ -247,7 +212,7 @@ class ChannelClient: return lines def _render(self, title, body_lines, accent=(60, 150, 255), footer=""): - w, h = self.board.LCD_WIDTH, self.board.LCD_HEIGHT + w, h = self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT img = Image.new("RGB", (w, h), (11, 16, 24)) draw = ImageDraw.Draw(img) draw.rounded_rectangle((12, 16, w - 12, h - 16), radius=18, @@ -283,13 +248,13 @@ class ChannelClient: return self._rgb565(img) def _flash_screen(self): - if self.board is None: + if self.whisplay_interface.board is None: return fb = self._render("LXST Client", ["Channels loading..."], footer="Please wait") - self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) + self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_channels(self): - if self.board is None: + if self.whisplay_interface.board is None: return if not self.channels: fb = self._render("LXST Client", ["No channels", "", "Re-discover to find channels"], @@ -300,10 +265,10 @@ class ChannelClient: lines.append((f"{i+1}: {ch['hash'][:16]}..", i == self.selected_index)) fb = self._render(f"Channels ({len(self.channels)})", lines, footer="Short: next | Long: select") - self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) + self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_in_call(self): - if self.board is None: + if self.whisplay_interface.board is None: return ch = self.channels[self.selected_index] if self.channels else None lines = [f"Channel {self.selected_index + 1}"] @@ -316,67 +281,66 @@ class ChannelClient: lines += ["", "Always-on (unmuted)"] footer = "Short: PTT Dbl: leave" fb = self._render("In Call", lines, accent=(60, 220, 120), footer=footer) - self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) + self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_status(self, title, detail="", accent=(255, 180, 60)): - if self.board is None: + if self.whisplay_interface.board is None: return lines = [detail] if detail else [] fb = self._render(title, lines, accent=accent) - self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) + self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) - def _on_button_press(self): - if self.state == self.STATE_IN_CALL: - if self.gestures.is_double_pressed: + def _gesture_loop(self): + while self.should_run: + self.whisplay_interface.update_gestures() + + # Double-press (highest priority) + if self.whisplay_interface.was_double_pressed and self.state == self.STATE_IN_CALL: self.telephone.hangup() - return - if self._ptt_enabled: - self._ptt_active = True - self.telephone.unmute_transmit() - if self.board: - self.board.set_rgb(255, 255, 0) + time.sleep(0.02) + continue - def _on_button_release(self): - if self.gestures.was_double_pressed: - self.gestures.acknowledge_double_press() - return - if self.state == self.STATE_IN_CALL: - elapsed = self.gestures.press_duration - if self._ptt_enabled: - if elapsed < self.gestures.long_press_threshold: - self._ptt_active = False - self._ptt_enabled = False - if self.board: - self._show_in_call() - else: - self._ptt_active = False - self.telephone.mute_transmit() - if self.board: - self.board.set_rgb(0, 255, 80) - else: - if elapsed < self.gestures.long_press_threshold: - self._ptt_enabled = True - self.telephone.mute_transmit() - if self.board: - self._show_in_call() - return - if self.state != self.STATE_READY: - return - elapsed = self.gestures.press_duration - if elapsed >= self.gestures.long_press_threshold: - if self.channels: - self.board.set_rgb(255, 180, 0) - self.call_channel(self.selected_index) - else: - if self.channels: - self.selected_index = (self.selected_index + 1) % len(self.channels) - self._show_channels() - self.gestures.acknowledge_press() + # STATE_READY: release actions + if self.state == self.STATE_READY and self.whisplay_interface.was_pressed: + if self.whisplay_interface.was_long_pressed and self.channels: + self.whisplay_interface.board.set_rgb(255, 180, 0) + self.call_channel(self.selected_index) + elif self.channels: + self.selected_index = (self.selected_index + 1) % len(self.channels) + self._show_channels() + + # STATE_IN_CALL: hold-to-talk + release actions + if self.state == self.STATE_IN_CALL: + if self.whisplay_interface.is_pressed and self._ptt_enabled and not self._ptt_active: + self._ptt_active = True + self.telephone.unmute_transmit() + if self.whisplay_interface.board: + self.whisplay_interface.board.set_rgb(255, 255, 0) + + if self.whisplay_interface.was_pressed: + if self._ptt_enabled: + if not self.whisplay_interface.was_long_pressed: + self._ptt_active = False + self._ptt_enabled = False + if self.whisplay_interface.board: + self._show_in_call() + else: + self._ptt_active = False + self.telephone.mute_transmit() + if self.whisplay_interface.board: + self.whisplay_interface.board.set_rgb(0, 255, 80) + elif not self.whisplay_interface.was_long_pressed: + self._ptt_enabled = True + self.telephone.mute_transmit() + if self.whisplay_interface.board: + self._show_in_call() + + time.sleep(0.02) def _on_exit_request(self, payload=None): self.cleanup() - if self.board: - self.board.prepare_exit() + if self.whisplay_interface.board: + self.whisplay_interface.board.prepare_exit() def _on_focus_revoked(self, payload=None): self.cleanup() @@ -498,8 +462,8 @@ class ChannelClient: ) if self._ptt_enabled: self.telephone.mute_transmit() - if self.board: - self.board.set_rgb(0, 255, 80) + if self.whisplay_interface.board: + self.whisplay_interface.board.set_rgb(0, 255, 80) self._show_in_call() def _on_call_ended(self, remote_identity): @@ -507,14 +471,12 @@ class ChannelClient: was_calling = self.state == self.STATE_CALLING self.state = self.STATE_READY self._ptt_active = False - if self.gestures: - self.gestures.acknowledge_double_press() if was_in_call: print(f"\nCall ended") elif was_calling: print(f"\nCall could not be connected") - if self.board: - self.board.set_rgb(0, 80, 255) + if self.whisplay_interface.board: + self.whisplay_interface.board.set_rgb(0, 80, 255) if was_in_call: self._show_status("Call Ended") time.sleep(1) @@ -593,7 +555,7 @@ class ChannelClient: signal.signal(signal.SIGTERM, self._sigterm_handler) self.should_run = True - if self.board: + if self.whisplay_interface.board: self._run_gui() else: self._run_cli() @@ -602,8 +564,10 @@ class ChannelClient: self._flash_screen() self.discover(timeout=15) self.selected_index = 0 - self.board.set_rgb(0, 80, 255) + self.whisplay_interface.board.set_rgb(0, 80, 255) self._show_channels() + self._gesture_thread = threading.Thread(target=self._gesture_loop, daemon=True) + self._gesture_thread.start() while self.should_run: if self.state == self.STATE_CALLING: self._show_status("Calling...", f"Channel {self.selected_index + 1}", @@ -679,10 +643,10 @@ class ChannelClient: if self.telephone and self.state == self.STATE_IN_CALL: self.telephone.hangup() self.should_run = False - if self.gestures: - self.gestures.stop() - if self.board: - self.board.cleanup() + if self._gesture_thread: + self._gesture_thread.join(timeout=1) + if self.whisplay_interface.board: + self.whisplay_interface.board.cleanup() def _sigint_handler(self, sig, frame): self.cleanup() diff --git a/runtime/whisplay_client.py b/runtime/whisplay_client.py index 757b38d..9e77a58 100644 --- a/runtime/whisplay_client.py +++ b/runtime/whisplay_client.py @@ -4,7 +4,7 @@ import socket import threading import time -from whisplay import WhisplayBoard +from runtime.whisplay import WhisplayBoard DEFAULT_DAEMON_SOCKET_PATH = "/tmp/whisplay-daemon.sock" diff --git a/whisplay_interface.py b/whisplay_interface.py index dd9b6e6..90cf359 100644 --- a/whisplay_interface.py +++ b/whisplay_interface.py @@ -1,38 +1,28 @@ import time import threading +import runtime.whisplay_client class WhisplayInterface: - def __init__(self, board=None, long_press_threshold=0.5, double_press_threshold=0.3): - self._board = board - self._long_press_threshold = long_press_threshold - self._double_press_threshold = double_press_threshold - self._button_is_double_pressed_set = False - + LONG_PRESS_THRESHOLD = 0.5 + DOUBLE_PRESS_THRESHOLD = 0.3 + + def __init__(self): + self.board = None + self._lock = threading.Lock() self._button_press_start = None self._button_press_end = None self._button_is_pressed = False self._button_was_pressed = False self._button_is_long_pressed = False self._button_was_long_pressed = False - self._button_is_double_pressed = False self._button_was_double_pressed = False - self._press_callbacks = [] - self._release_callbacks = [] - self._running = False - self._gesture_thread = None + # Edge-detection state + self._prev_is_pressed = False + self._prev_press_end = None - if board is not None: - self.attach(board) - - @property - def long_press_threshold(self): - return self._long_press_threshold - - @property - def double_press_threshold(self): - return self._double_press_threshold + # -- Properties -- @property def is_pressed(self): @@ -50,10 +40,6 @@ class WhisplayInterface: def was_long_pressed(self): return self._button_was_long_pressed - @property - def is_double_pressed(self): - return self._button_is_double_pressed - @property def was_double_pressed(self): return self._button_was_double_pressed @@ -65,76 +51,63 @@ class WhisplayInterface: end = self._button_press_end if self._button_press_end is not None else time.time() return end - self._button_press_start - def attach(self, board): - self._board = board - self._board.on_button_press(self._on_button_press) - self._board.on_button_release(self._on_button_release) - self.start() + # -- Attachment -- - def start(self): - if self._gesture_thread is not None: - return - self._running = True - self._gesture_thread = threading.Thread(daemon=True, target=self._gesture_loop) - self._gesture_thread.start() + def create_board(self, + app_id, + display_name, + icon, + use_daemon_default_log=True): + self.board = runtime.whisplay_client.create_whisplay_hardware(app_id=app_id, display_name=display_name, icon=icon, use_daemon_default_log=use_daemon_default_log) + self.board.on_button_press(self._on_button_press) + self.board.on_button_release(self._on_button_release) - def stop(self): - self._running = False - - def on_press(self, callback): - self._press_callbacks.append(callback) - - def on_release(self, callback): - self._release_callbacks.append(callback) - - def acknowledge_press(self): - self._button_was_pressed = False - - def acknowledge_long_press(self): - self._button_was_long_pressed = False - - def acknowledge_double_press(self): - self._button_was_double_pressed = False + # -- Internal press/release (called from GPIO thread) -- def _on_button_press(self): - self._button_is_pressed = True - self._button_press_start = time.time() - if self._button_press_end is not None: - gap = self._button_press_start - self._button_press_end - if gap < self._double_press_threshold: - self._button_is_double_pressed = True - self._button_is_double_pressed_set = True - for cb in self._press_callbacks: - cb() + with self._lock: + self._button_is_pressed = True + self._button_press_start = time.time() def _on_button_release(self): - self._button_was_pressed = self._button_is_pressed - self._button_press_end = time.time() - self._button_was_long_pressed = self._button_is_long_pressed - self._button_was_double_pressed = self._button_is_double_pressed - self._button_is_pressed = False - self._button_is_double_pressed = False - self._button_is_long_pressed = False - self._button_is_double_pressed_set = False - for cb in self._release_callbacks: - cb() + with self._lock: + self._button_is_pressed = False + self._button_press_end = time.time() - def _gesture_loop(self): - while self._running: - try: - current_time = time.time() - if self._button_is_pressed: - if not self._button_is_double_pressed_set: - self._button_is_double_pressed = ( - self._button_was_pressed - and self._button_press_end is not None - and current_time - self._button_press_end < self._double_press_threshold - ) - if self._button_is_double_pressed: - self._button_is_double_pressed_set = True - self._button_is_long_pressed = ( - current_time - self._button_press_start > self._long_press_threshold - ) - except Exception: - pass - time.sleep(0.05) + # -- Polled update (called from gesture thread) -- + + def update_gestures(self): + self._button_was_pressed = False + self._button_was_long_pressed = False + self._button_was_double_pressed = False + + with self._lock: + is_pressed = self._button_is_pressed + press_start = self._button_press_start + press_end = self._button_press_end + + now = time.time() + + # Press edge: transitioned from not-pressed to pressed + if not self._prev_is_pressed and is_pressed: + if self._prev_press_end is not None and press_start is not None: + gap = press_start - self._prev_press_end + if gap < WhisplayInterface.DOUBLE_PRESS_THRESHOLD: + self._button_was_double_pressed = True + + # Release edge: transitioned from pressed to not-pressed + if self._prev_is_pressed and not is_pressed: + self._button_was_pressed = True + if press_start is not None and press_end is not None: + duration = press_end - press_start + if duration >= WhisplayInterface.LONG_PRESS_THRESHOLD: + self._button_was_long_pressed = True + self._prev_press_end = press_end + + # Level flag: currently held past threshold + if is_pressed and press_start is not None: + self._button_is_long_pressed = (now - press_start) >= WhisplayInterface.LONG_PRESS_THRESHOLD + else: + self._button_is_long_pressed = False + + self._prev_is_pressed = is_pressed