Compare commits
6 Commits
feed1145f8
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 9e888d9afa | |||
| 884325d1c6 | |||
| 3b5bac5d2c | |||
| 63422c9a75 | |||
| 561c014c33 | |||
| b4cd86ed78 |
@@ -1,8 +1,8 @@
|
|||||||
# LXST Client
|
# LXST Client
|
||||||
|
|
||||||
Single-file Python application for managing LXST Channel connections via Reticulum Network (RNS) and LXMF protocols. Supports CLI and Whisplay HAT GUI modes.
|
Single-file Python application for managing LXST Channel connections via Reticulum Network (RNS) and LXMF protocols using the Whisplay HAT.
|
||||||
|
|
||||||
## Running (CLI mode, default)
|
## Running
|
||||||
|
|
||||||
python client.py [-c CONFIGDIR] [-w WHITELIST]
|
python client.py [-c CONFIGDIR] [-w WHITELIST]
|
||||||
|
|
||||||
@@ -11,20 +11,6 @@ Single-file Python application for managing LXST Channel connections via Reticul
|
|||||||
-w, --whitelist : path to bridge_whitelist.json
|
-w, --whitelist : path to bridge_whitelist.json
|
||||||
-v, --verbose : enable debug logging
|
-v, --verbose : enable debug logging
|
||||||
--rnsconfig : alternative Reticulum config directory
|
--rnsconfig : alternative Reticulum config directory
|
||||||
--whisplay, -W : enable Whisplay HAT GUI mode
|
|
||||||
|
|
||||||
## Running (Whisplay GUI mode)
|
|
||||||
|
|
||||||
python client.py -W [-c CONFIGDIR] [-w WHITELIST]
|
|
||||||
|
|
||||||
## CLI Commands
|
|
||||||
|
|
||||||
Inside the interactive client:
|
|
||||||
d Discover channels from whitelist
|
|
||||||
l List discovered channels
|
|
||||||
<NUM> Connect to channel at index NUM (1-based)
|
|
||||||
h / ? Show help
|
|
||||||
q Quit
|
|
||||||
|
|
||||||
## Whisplay GUI Controls
|
## Whisplay GUI Controls
|
||||||
|
|
||||||
@@ -56,9 +42,8 @@ Single-file Python application for managing LXST Channel connections via Reticul
|
|||||||
1. Ensure ~/.lxst-client or /etc/lxst-client exists
|
1. Ensure ~/.lxst-client or /etc/lxst-client exists
|
||||||
2. Place bridge_whitelist.json in either directory (or pass with -w)
|
2. Place bridge_whitelist.json in either directory (or pass with -w)
|
||||||
3. First run creates identity automatically at configdir/identity
|
3. First run creates identity automatically at configdir/identity
|
||||||
4. In CLI mode: run `python client.py` then `d` to discover channels
|
4. In GUI mode: run `python client.py -W` — auto-discovers on startup
|
||||||
5. In GUI mode: run `python client.py -W` — auto-discovers on startup
|
5. Messages appear in configdir/lxmf/
|
||||||
6. Messages appear in configdir/lxmf/
|
|
||||||
|
|
||||||
## Known quirks
|
## Known quirks
|
||||||
|
|
||||||
@@ -67,4 +52,3 @@ Single-file Python application for managing LXST Channel connections via Reticul
|
|||||||
- Use -v for RNS debug output (path discovery is verbose here).
|
- Use -v for RNS debug output (path discovery is verbose here).
|
||||||
- Path discovery waits up to 10s for relays; increase timeout in client code if needed.
|
- Path discovery waits up to 10s for relays; increase timeout in client code if needed.
|
||||||
- Whisplay mode requires Pillow and the Whisplay runtime (whisplay.py / whisplay_client.py).
|
- Whisplay mode requires Pillow and the Whisplay runtime (whisplay.py / whisplay_client.py).
|
||||||
- Falls back to CLI gracefully if Whisplay hardware or dependencies are unavailable.
|
|
||||||
|
|||||||
@@ -1,3 +1,3 @@
|
|||||||
[
|
[
|
||||||
"8567367936f085f93bf4ecdf494fafd0"
|
"8567367936f085f93bf4ecdf494fafd0", "c64bc5a948096bd624bd9c186c35bdb1"
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ class ChannelClient:
|
|||||||
STATE_IN_CALL = 4
|
STATE_IN_CALL = 4
|
||||||
|
|
||||||
def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None,
|
def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None,
|
||||||
verbosity=0, announce_interval=0, use_whisplay=False):
|
verbosity=0, announce_interval=0):
|
||||||
self.state = self.STATE_IDLE
|
self.state = self.STATE_IDLE
|
||||||
self.channels = []
|
self.channels = []
|
||||||
self.telephone = None
|
self.telephone = None
|
||||||
@@ -50,10 +50,8 @@ 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
|
|
||||||
|
|
||||||
if use_whisplay:
|
self._init_whisplay()
|
||||||
self._init_whisplay()
|
|
||||||
|
|
||||||
if configdir is None:
|
if configdir is None:
|
||||||
if os.path.isdir("/etc/lxst-client"):
|
if os.path.isdir("/etc/lxst-client"):
|
||||||
@@ -118,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:
|
||||||
@@ -275,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)
|
||||||
@@ -290,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()
|
||||||
@@ -461,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()
|
||||||
@@ -482,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")
|
||||||
@@ -555,10 +583,7 @@ class ChannelClient:
|
|||||||
signal.signal(signal.SIGTERM, self._sigterm_handler)
|
signal.signal(signal.SIGTERM, self._sigterm_handler)
|
||||||
self.should_run = True
|
self.should_run = True
|
||||||
|
|
||||||
if self.whisplay_interface.board:
|
self._run_gui()
|
||||||
self._run_gui()
|
|
||||||
else:
|
|
||||||
self._run_cli()
|
|
||||||
|
|
||||||
def _run_gui(self):
|
def _run_gui(self):
|
||||||
self._flash_screen()
|
self._flash_screen()
|
||||||
@@ -566,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,75 +599,10 @@ class ChannelClient:
|
|||||||
continue
|
continue
|
||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
|
|
||||||
def _run_cli(self):
|
|
||||||
print(
|
|
||||||
f"\n{Terminal.BOLD}"
|
|
||||||
f"LXST Channel Client"
|
|
||||||
f"{Terminal.END}"
|
|
||||||
)
|
|
||||||
print(f" Identity: {RNS.prettyhexrep(self.identity.hash)}")
|
|
||||||
print(f" Whitelist: {self.whitelist_path}")
|
|
||||||
print()
|
|
||||||
|
|
||||||
self.discover(timeout=15)
|
|
||||||
self.show_channels()
|
|
||||||
self.show_help()
|
|
||||||
|
|
||||||
while self.should_run:
|
|
||||||
if self.state == self.STATE_IN_CALL:
|
|
||||||
try:
|
|
||||||
input(" In call, press Enter to hang up > ")
|
|
||||||
self.telephone.hangup()
|
|
||||||
self.state = self.STATE_READY
|
|
||||||
except (EOFError, KeyboardInterrupt):
|
|
||||||
break
|
|
||||||
continue
|
|
||||||
|
|
||||||
if self.state == self.STATE_CALLING:
|
|
||||||
time.sleep(0.5)
|
|
||||||
continue
|
|
||||||
|
|
||||||
if self.state == self.STATE_DISCOVERING:
|
|
||||||
time.sleep(0.5)
|
|
||||||
continue
|
|
||||||
|
|
||||||
try:
|
|
||||||
inp = input("> ").strip().lower()
|
|
||||||
except (EOFError, KeyboardInterrupt):
|
|
||||||
print()
|
|
||||||
break
|
|
||||||
|
|
||||||
if not inp:
|
|
||||||
continue
|
|
||||||
|
|
||||||
if inp == "q":
|
|
||||||
break
|
|
||||||
elif inp in ("h", "?"):
|
|
||||||
self.show_help()
|
|
||||||
elif inp == "d":
|
|
||||||
self.discover()
|
|
||||||
self.show_channels()
|
|
||||||
elif inp == "l":
|
|
||||||
self.show_channels()
|
|
||||||
else:
|
|
||||||
try:
|
|
||||||
idx = int(inp) - 1
|
|
||||||
if 0 <= idx < len(self.channels):
|
|
||||||
self.call_channel(idx)
|
|
||||||
else:
|
|
||||||
print(
|
|
||||||
f"Invalid index. "
|
|
||||||
f"Enter a number between 1 and {len(self.channels)}"
|
|
||||||
)
|
|
||||||
except ValueError:
|
|
||||||
print(f"Unknown command: {inp}. Type 'h' for help.")
|
|
||||||
|
|
||||||
def cleanup(self):
|
def cleanup(self):
|
||||||
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()
|
||||||
|
|
||||||
@@ -682,11 +640,6 @@ def main():
|
|||||||
help="seconds between LXMF destination announces (0 = announce once at startup, default)",
|
help="seconds between LXMF destination announces (0 = announce once at startup, default)",
|
||||||
type=int,
|
type=int,
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
|
||||||
"--whisplay", "-W",
|
|
||||||
action="store_true",
|
|
||||||
help="enable Whisplay HAT GUI mode",
|
|
||||||
)
|
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
client = ChannelClient(
|
client = ChannelClient(
|
||||||
@@ -695,7 +648,6 @@ def main():
|
|||||||
whitelist_path=args.whitelist,
|
whitelist_path=args.whitelist,
|
||||||
verbosity=args.verbose,
|
verbosity=args.verbose,
|
||||||
announce_interval=args.announce_interval,
|
announce_interval=args.announce_interval,
|
||||||
use_whisplay=args.whisplay,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|||||||
+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