Compare commits
4 Commits
561c014c33
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 9e888d9afa | |||
| 884325d1c6 | |||
| 3b5bac5d2c | |||
| 63422c9a75 |
@@ -1,3 +1,3 @@
|
|||||||
[
|
[
|
||||||
"8567367936f085f93bf4ecdf494fafd0"
|
"8567367936f085f93bf4ecdf494fafd0", "c64bc5a948096bd624bd9c186c35bdb1"
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -50,7 +50,6 @@ class ChannelClient:
|
|||||||
self._ptt_enabled = True
|
self._ptt_enabled = True
|
||||||
self._ptt_active = False
|
self._ptt_active = False
|
||||||
self.whisplay_interface = None
|
self.whisplay_interface = None
|
||||||
self._gesture_thread = None
|
|
||||||
|
|
||||||
self._init_whisplay()
|
self._init_whisplay()
|
||||||
|
|
||||||
@@ -117,17 +116,20 @@ class ChannelClient:
|
|||||||
|
|
||||||
def _init_whisplay(self):
|
def _init_whisplay(self):
|
||||||
self.whisplay_interface = WhisplayInterface()
|
self.whisplay_interface = WhisplayInterface()
|
||||||
self.whisplay_interface.create_board(
|
self.whisplay_interface.create_board()
|
||||||
app_id="lxst-client",
|
|
||||||
display_name="LXST Client",
|
|
||||||
icon="LC",
|
|
||||||
use_daemon_default_log=True,
|
|
||||||
)
|
|
||||||
self.whisplay_interface.board.set_backlight(70)
|
self.whisplay_interface.board.set_backlight(70)
|
||||||
# self.whisplay_interface.board.on_exit_request(self._on_exit_request)
|
if hasattr(self.whisplay_interface.board, "on_exit_request"):
|
||||||
# self.whisplay_interface.board.on_focus_revoked(self._on_focus_revoked)
|
self.whisplay_interface.board.on_exit_request(self._on_exit_request)
|
||||||
|
if hasattr(self.whisplay_interface.board, "on_focus_revoked"):
|
||||||
|
self.whisplay_interface.board.on_focus_revoked(self._on_focus_revoked)
|
||||||
self._load_fonts()
|
self._load_fonts()
|
||||||
self._setup_whisplay_audio()
|
self._setup_whisplay_audio()
|
||||||
|
self.whisplay_interface.on_press(self._on_press)
|
||||||
|
self.whisplay_interface.on_release(self._on_release)
|
||||||
|
self.whisplay_interface.on_long_press(self._on_long_press)
|
||||||
|
self.whisplay_interface.on_long_release(self._on_long_release)
|
||||||
|
self.whisplay_interface.on_double_press(self._on_double_press)
|
||||||
|
self.whisplay_interface.on_double_release(self._on_double_release)
|
||||||
|
|
||||||
def _find_wm8960_card(self):
|
def _find_wm8960_card(self):
|
||||||
try:
|
try:
|
||||||
@@ -274,9 +276,9 @@ class ChannelClient:
|
|||||||
if ch:
|
if ch:
|
||||||
lines.append(ch["hash"][:16] + "..")
|
lines.append(ch["hash"][:16] + "..")
|
||||||
if self._ptt_enabled:
|
if self._ptt_enabled:
|
||||||
lines += ["", "PTT: hold to talk", "Short: always-on"]
|
lines += ["", "PTT: hold to talk", "release to mute"]
|
||||||
footer = "Dbl: leave"
|
footer = "Short: always-on | Dbl: leave"
|
||||||
else:
|
else:
|
||||||
lines += ["", "Always-on (unmuted)"]
|
lines += ["", "Always-on (unmuted)"]
|
||||||
footer = "Short: PTT Dbl: leave"
|
footer = "Short: PTT Dbl: leave"
|
||||||
fb = self._render("In Call", lines, accent=(60, 220, 120), footer=footer)
|
fb = self._render("In Call", lines, accent=(60, 220, 120), footer=footer)
|
||||||
@@ -289,52 +291,66 @@ class ChannelClient:
|
|||||||
fb = self._render(title, lines, accent=accent)
|
fb = self._render(title, lines, accent=accent)
|
||||||
self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.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 _gesture_loop(self):
|
def _on_press(self):
|
||||||
while self.should_run:
|
if self.state == self.STATE_IN_CALL:
|
||||||
self.whisplay_interface.update_gestures()
|
if self._ptt_enabled and not self._ptt_active:
|
||||||
|
self._ptt_active = True
|
||||||
|
self._hw_set_mic_muted(False)
|
||||||
|
if self.whisplay_interface.board:
|
||||||
|
self.whisplay_interface.board.set_rgb(255, 0, 0)
|
||||||
|
|
||||||
# Double-press (highest priority)
|
def _on_double_press(self):
|
||||||
if self.whisplay_interface.was_double_pressed and self.state == self.STATE_IN_CALL:
|
if self.state == self.STATE_READY:
|
||||||
self.telephone.hangup()
|
self._on_press()
|
||||||
time.sleep(0.02)
|
|
||||||
continue
|
|
||||||
|
|
||||||
# STATE_READY: release actions
|
def _on_long_press(self):
|
||||||
if self.state == self.STATE_READY and self.whisplay_interface.was_pressed:
|
pass
|
||||||
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
|
def _on_release(self):
|
||||||
if self.state == self.STATE_IN_CALL:
|
if self.state == self.STATE_READY:
|
||||||
if self.whisplay_interface.is_pressed and self._ptt_enabled and not self._ptt_active:
|
if self.channels:
|
||||||
self._ptt_active = True
|
self.selected_index = (self.selected_index + 1) % len(self.channels)
|
||||||
self.telephone.unmute_transmit()
|
self._show_channels()
|
||||||
if self.whisplay_interface.board:
|
|
||||||
self.whisplay_interface.board.set_rgb(255, 255, 0)
|
|
||||||
|
|
||||||
if self.whisplay_interface.was_pressed:
|
elif self.state == self.STATE_IN_CALL:
|
||||||
if self._ptt_enabled:
|
if self._ptt_enabled:
|
||||||
if not self.whisplay_interface.was_long_pressed:
|
self._ptt_enabled = False
|
||||||
self._ptt_active = False
|
self._ptt_active = False
|
||||||
self._ptt_enabled = False
|
if self.whisplay_interface.board:
|
||||||
if self.whisplay_interface.board:
|
self.whisplay_interface.board.set_rgb(255, 0, 0)
|
||||||
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)
|
else:
|
||||||
|
self._ptt_enabled = True
|
||||||
|
self._ptt_active = False
|
||||||
|
self._hw_set_mic_muted(True)
|
||||||
|
if self.whisplay_interface.board:
|
||||||
|
self.whisplay_interface.board.set_rgb(0, 0, 0)
|
||||||
|
|
||||||
|
if self.whisplay_interface.board:
|
||||||
|
self._show_in_call()
|
||||||
|
|
||||||
|
def _on_double_release(self):
|
||||||
|
if self.state == self.STATE_IN_CALL:
|
||||||
|
self.telephone.hangup()
|
||||||
|
self.selected_index = (self.selected_index + 1) % len(self.channels)
|
||||||
|
self._show_channels()
|
||||||
|
|
||||||
|
def _on_long_release(self):
|
||||||
|
if self.state == self.STATE_IN_CALL and self._ptt_enabled:
|
||||||
|
self._ptt_active = False
|
||||||
|
self._hw_set_mic_muted(True)
|
||||||
|
if self.whisplay_interface.board:
|
||||||
|
self.whisplay_interface.board.set_rgb(0, 0, 0)
|
||||||
|
elif self.state == self.STATE_READY:
|
||||||
|
if self.channels:
|
||||||
|
if self._ptt_enabled:
|
||||||
|
self.whisplay_interface.board.set_rgb(0, 0, 0)
|
||||||
|
else:
|
||||||
|
self.whisplay_interface.board.set_rgb(255, 0, 0)
|
||||||
|
self.call_channel(self.selected_index)
|
||||||
|
|
||||||
|
else:
|
||||||
|
self._on_release()
|
||||||
|
|
||||||
def _on_exit_request(self, payload=None):
|
def _on_exit_request(self, payload=None):
|
||||||
self.cleanup()
|
self.cleanup()
|
||||||
@@ -460,7 +476,7 @@ class ChannelClient:
|
|||||||
f"{RNS.prettyhexrep(remote_identity.hash)}"
|
f"{RNS.prettyhexrep(remote_identity.hash)}"
|
||||||
)
|
)
|
||||||
if self._ptt_enabled:
|
if self._ptt_enabled:
|
||||||
self.telephone.mute_transmit()
|
self._hw_set_mic_muted(True)
|
||||||
if self.whisplay_interface.board:
|
if self.whisplay_interface.board:
|
||||||
self.whisplay_interface.board.set_rgb(0, 255, 80)
|
self.whisplay_interface.board.set_rgb(0, 255, 80)
|
||||||
self._show_in_call()
|
self._show_in_call()
|
||||||
@@ -481,6 +497,19 @@ class ChannelClient:
|
|||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
self._show_channels()
|
self._show_channels()
|
||||||
|
|
||||||
|
def _hw_set_mic_muted(self, muted):
|
||||||
|
if self._wm8960_device is None:
|
||||||
|
return
|
||||||
|
card = self._wm8960_device.split(",")[0].split(":")[-1]
|
||||||
|
state = "off" if muted else "on"
|
||||||
|
try:
|
||||||
|
subprocess.run(
|
||||||
|
["amixer", "-c", card, "cset", "name='Capture Switch'", state],
|
||||||
|
check=False, capture_output=True, timeout=2,
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
def call_channel(self, index):
|
def call_channel(self, index):
|
||||||
if index < 0 or index >= len(self.channels):
|
if index < 0 or index >= len(self.channels):
|
||||||
print("Invalid channel index")
|
print("Invalid channel index")
|
||||||
@@ -562,8 +591,6 @@ class ChannelClient:
|
|||||||
self.selected_index = 0
|
self.selected_index = 0
|
||||||
self.whisplay_interface.board.set_rgb(0, 80, 255)
|
self.whisplay_interface.board.set_rgb(0, 80, 255)
|
||||||
self._show_channels()
|
self._show_channels()
|
||||||
self._gesture_thread = threading.Thread(target=self._gesture_loop, daemon=True)
|
|
||||||
self._gesture_thread.start()
|
|
||||||
while self.should_run:
|
while self.should_run:
|
||||||
if self.state == self.STATE_CALLING:
|
if self.state == self.STATE_CALLING:
|
||||||
self._show_status("Calling...", f"Channel {self.selected_index + 1}",
|
self._show_status("Calling...", f"Channel {self.selected_index + 1}",
|
||||||
@@ -576,8 +603,6 @@ class ChannelClient:
|
|||||||
if self.telephone and self.state == self.STATE_IN_CALL:
|
if self.telephone and self.state == self.STATE_IN_CALL:
|
||||||
self.telephone.hangup()
|
self.telephone.hangup()
|
||||||
self.should_run = False
|
self.should_run = False
|
||||||
if self._gesture_thread:
|
|
||||||
self._gesture_thread.join(timeout=1)
|
|
||||||
if self.whisplay_interface.board:
|
if self.whisplay_interface.board:
|
||||||
self.whisplay_interface.board.cleanup()
|
self.whisplay_interface.board.cleanup()
|
||||||
|
|
||||||
|
|||||||
+81
-81
@@ -1,5 +1,7 @@
|
|||||||
|
import subprocess
|
||||||
import time
|
import time
|
||||||
import threading
|
import threading
|
||||||
|
import os
|
||||||
import runtime.whisplay_client
|
import runtime.whisplay_client
|
||||||
|
|
||||||
|
|
||||||
@@ -10,104 +12,102 @@ class WhisplayInterface:
|
|||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.board = None
|
self.board = None
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
self._is_pressing = False
|
||||||
|
self._is_double_pressing = False
|
||||||
|
self._is_long_pressing = False
|
||||||
|
self._long_press_timer_thread = None
|
||||||
self._button_press_start = None
|
self._button_press_start = None
|
||||||
self._button_press_end = None
|
self._button_press_end = None
|
||||||
self._button_is_pressed = False
|
self._on_press_callback = None
|
||||||
self._button_was_pressed = False
|
self._on_release_callback = None
|
||||||
self._button_is_long_pressed = False
|
self._on_long_press_callback = None
|
||||||
self._button_was_long_pressed = False
|
self._on_long_release_callback = None
|
||||||
self._button_was_double_pressed = False
|
self._on_double_press_callback = None
|
||||||
|
self._on_double_release_callback = None
|
||||||
|
|
||||||
# Edge-detection state
|
self.set_wm8960_volume_stable(speaker=200, mic=80)
|
||||||
self._prev_is_pressed = False
|
|
||||||
self._prev_press_end = None
|
|
||||||
|
|
||||||
# -- Properties --
|
def create_board(self):
|
||||||
|
self.board = runtime.whisplay_client.create_whisplay_hardware(
|
||||||
@property
|
app_id=os.getenv("WHISPLAY_APP_ID", "whisplay-lxst-client"),
|
||||||
def is_pressed(self):
|
display_name="LXST Client",
|
||||||
return self._button_is_pressed
|
icon="L",
|
||||||
|
exit_gesture="quad_press",
|
||||||
@property
|
use_daemon_default_log=False,
|
||||||
def was_pressed(self):
|
)
|
||||||
return self._button_was_pressed
|
|
||||||
|
|
||||||
@property
|
|
||||||
def is_long_pressed(self):
|
|
||||||
return self._button_is_long_pressed
|
|
||||||
|
|
||||||
@property
|
|
||||||
def was_long_pressed(self):
|
|
||||||
return self._button_was_long_pressed
|
|
||||||
|
|
||||||
@property
|
|
||||||
def was_double_pressed(self):
|
|
||||||
return self._button_was_double_pressed
|
|
||||||
|
|
||||||
@property
|
|
||||||
def press_duration(self):
|
|
||||||
if self._button_press_start is None:
|
|
||||||
return 0.0
|
|
||||||
end = self._button_press_end if self._button_press_end is not None else time.time()
|
|
||||||
return end - self._button_press_start
|
|
||||||
|
|
||||||
# -- Attachment --
|
|
||||||
|
|
||||||
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_press(self._on_button_press)
|
||||||
self.board.on_button_release(self._on_button_release)
|
self.board.on_button_release(self._on_button_release)
|
||||||
|
|
||||||
# -- Internal press/release (called from GPIO thread) --
|
# -- Internal press/release (called from GPIO thread) --
|
||||||
|
|
||||||
|
def on_press(self, callback):
|
||||||
|
self._on_press_callback = callback
|
||||||
|
|
||||||
|
def on_release(self, callback):
|
||||||
|
self._on_release_callback = callback
|
||||||
|
|
||||||
|
def on_long_press(self, callback):
|
||||||
|
self._on_long_press_callback = callback
|
||||||
|
|
||||||
|
def on_long_release(self, callback):
|
||||||
|
self._on_long_release_callback = callback
|
||||||
|
|
||||||
|
def on_double_press(self, callback):
|
||||||
|
self._on_double_press_callback = callback
|
||||||
|
|
||||||
|
def on_double_release(self, callback):
|
||||||
|
self._on_double_release_callback = callback
|
||||||
|
|
||||||
|
|
||||||
def _on_button_press(self):
|
def _on_button_press(self):
|
||||||
with self._lock:
|
with self._lock:
|
||||||
self._button_is_pressed = True
|
self._is_pressing = True
|
||||||
self._button_press_start = time.time()
|
self._button_press_start = time.time()
|
||||||
|
if self._button_press_end is not None:
|
||||||
|
time_since_last_press = self._button_press_start- self._button_press_end
|
||||||
|
if time_since_last_press <= self.DOUBLE_PRESS_THRESHOLD:
|
||||||
|
self._is_double_pressing = True
|
||||||
|
if self._on_double_press_callback:
|
||||||
|
self._on_double_press_callback()
|
||||||
|
if not self._is_double_pressing:
|
||||||
|
if self._on_press_callback:
|
||||||
|
self._on_press_callback()
|
||||||
|
self._long_press_timer_thread = threading.Timer(self.LONG_PRESS_THRESHOLD, self._long_press_timer)
|
||||||
|
self._long_press_timer_thread.start()
|
||||||
|
|
||||||
|
|
||||||
def _on_button_release(self):
|
def _on_button_release(self):
|
||||||
with self._lock:
|
with self._lock:
|
||||||
self._button_is_pressed = False
|
self._long_press_timer_thread.cancel()
|
||||||
|
self._is_pressing = False
|
||||||
self._button_press_end = time.time()
|
self._button_press_end = time.time()
|
||||||
|
if self._is_double_pressing:
|
||||||
|
if self._on_double_release_callback:
|
||||||
|
self._on_double_release_callback()
|
||||||
|
self._is_double_pressing = False
|
||||||
|
elif self._is_long_pressing:
|
||||||
|
if self._on_long_release_callback:
|
||||||
|
self._on_long_release_callback()
|
||||||
|
self._is_long_pressing = False
|
||||||
|
else:
|
||||||
|
if self._on_release_callback:
|
||||||
|
self._on_release_callback()
|
||||||
|
|
||||||
# -- Polled update (called from gesture thread) --
|
def _long_press_timer(self):
|
||||||
|
|
||||||
def update_gestures(self):
|
|
||||||
self._button_was_pressed = False
|
|
||||||
self._button_was_long_pressed = False
|
|
||||||
self._button_was_double_pressed = False
|
|
||||||
|
|
||||||
with self._lock:
|
with self._lock:
|
||||||
is_pressed = self._button_is_pressed
|
if self._is_pressing:
|
||||||
press_start = self._button_press_start
|
self._is_long_pressing = True
|
||||||
press_end = self._button_press_end
|
if self._on_long_press_callback:
|
||||||
|
self._on_long_press_callback()
|
||||||
|
|
||||||
now = time.time()
|
|
||||||
|
|
||||||
# Press edge: transitioned from not-pressed to pressed
|
def set_wm8960_volume_stable(self, speaker: int = 0, mic: int = 0):
|
||||||
if not self._prev_is_pressed and is_pressed:
|
CARD_NAME = 'wm8960soundcard'
|
||||||
if self._prev_press_end is not None and press_start is not None:
|
DEVICE_ARG = f'hw:{CARD_NAME}'
|
||||||
gap = press_start - self._prev_press_end
|
try:
|
||||||
if gap < WhisplayInterface.DOUBLE_PRESS_THRESHOLD:
|
subprocess.run(['amixer', '-D', DEVICE_ARG, 'sset', 'Speaker',
|
||||||
self._button_was_double_pressed = True
|
str(speaker)], check=False, capture_output=True)
|
||||||
|
subprocess.run(['amixer', '-D', DEVICE_ARG, 'sset',
|
||||||
# Release edge: transitioned from pressed to not-pressed
|
'Capture', str(mic)], check=False, capture_output=True)
|
||||||
if self._prev_is_pressed and not is_pressed:
|
except Exception as e:
|
||||||
self._button_was_pressed = True
|
print(f"ERROR: Failed to set volume: {e}")
|
||||||
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
|
|
||||||
|
|||||||
Reference in New Issue
Block a user