whisplay renders list of channels, selects channel, and connects
This commit is contained in:
@@ -1,34 +1,50 @@
|
|||||||
# LXST Client
|
# LXST Client
|
||||||
|
|
||||||
Single-file Python CLI for managing LXST Channel connections via Reticulum Network (RNS) and LXMF protocols.
|
Single-file Python application for managing LXST Channel connections via Reticulum Network (RNS) and LXMF protocols. Supports CLI and Whisplay HAT GUI modes.
|
||||||
|
|
||||||
## Running
|
## Running (CLI mode, default)
|
||||||
|
|
||||||
python client.py [-c CONFIGDIR] [-w WHITELIST]
|
python client.py [-c CONFIGDIR] [-w WHITELIST]
|
||||||
|
|
||||||
Options:
|
Options:
|
||||||
-c, --config : config directory (default: ~/.lxst-client or /etc/lxst-client)
|
-c, --config : config directory (default: ~/.lxst-client or /etc/lxst-client)
|
||||||
-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
|
||||||
|
|
||||||
## Commands
|
## Running (Whisplay GUI mode)
|
||||||
|
|
||||||
python client.py -c /etc/lxst-client -w /etc/lxst-client/bridge_whitelist.json
|
python client.py -W [-c CONFIGDIR] [-w WHITELIST]
|
||||||
|
|
||||||
|
## CLI Commands
|
||||||
|
|
||||||
Inside the interactive client:
|
Inside the interactive client:
|
||||||
d Discover channels from whitelist
|
d Discover channels from whitelist
|
||||||
l List discovered channels
|
l List discovered channels
|
||||||
<NUM> Connect to channel at index NUM (1-based)
|
<NUM> Connect to channel at index NUM (1-based)
|
||||||
|
h / ? Show help
|
||||||
q Quit
|
q Quit
|
||||||
|
|
||||||
|
## Whisplay GUI Controls
|
||||||
|
|
||||||
|
Short press (<1s) Cycle to next channel in list
|
||||||
|
Long press (>=1s) Call the selected channel
|
||||||
|
Press (while in call) Hang up
|
||||||
|
|
||||||
|
RGB LED indicates state:
|
||||||
|
Blue = ready, showing channels
|
||||||
|
Orange = calling
|
||||||
|
Green = in call
|
||||||
|
|
||||||
## Initial Setup Order
|
## Initial Setup Order
|
||||||
|
|
||||||
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. Run `python client.py` then `d` to discover channels
|
4. In CLI mode: run `python client.py` then `d` to discover channels
|
||||||
5. Messages appear in configdir/lxmf/
|
5. In GUI mode: run `python client.py -W` — auto-discovers on startup
|
||||||
|
6. Messages appear in configdir/lxmf/
|
||||||
|
|
||||||
## Known quirks
|
## Known quirks
|
||||||
|
|
||||||
@@ -36,3 +52,5 @@ Single-file Python CLI for managing LXST Channel connections via Reticulum Netwo
|
|||||||
- Identity is auto-generated on first run if not present.
|
- Identity is auto-generated on first run if not present.
|
||||||
- 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).
|
||||||
|
- Falls back to CLI gracefully if Whisplay hardware or dependencies are unavailable.
|
||||||
|
|||||||
@@ -13,6 +13,29 @@ import LXMF
|
|||||||
from LXST.Primitives.Telephony import Telephone
|
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
|
||||||
|
|
||||||
|
LONG_PRESS_THRESHOLD = 1.0
|
||||||
|
|
||||||
|
|
||||||
class Terminal:
|
class Terminal:
|
||||||
if not RNS.vendor.platformutils.is_windows():
|
if not RNS.vendor.platformutils.is_windows():
|
||||||
UNDERLINE = "\033[4m"
|
UNDERLINE = "\033[4m"
|
||||||
@@ -31,7 +54,8 @@ class ChannelClient:
|
|||||||
STATE_CALLING = 3
|
STATE_CALLING = 3
|
||||||
STATE_IN_CALL = 4
|
STATE_IN_CALL = 4
|
||||||
|
|
||||||
def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None, verbosity=0, announce_interval=0):
|
def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None,
|
||||||
|
verbosity=0, announce_interval=0, use_whisplay=False):
|
||||||
self.state = self.STATE_IDLE
|
self.state = self.STATE_IDLE
|
||||||
self.channels = []
|
self.channels = []
|
||||||
self.telephone = None
|
self.telephone = None
|
||||||
@@ -40,6 +64,17 @@ class ChannelClient:
|
|||||||
self._responses = []
|
self._responses = []
|
||||||
self._response_event = threading.Event()
|
self._response_event = threading.Event()
|
||||||
|
|
||||||
|
# ── Whisplay HAT init ──
|
||||||
|
self.board = None
|
||||||
|
self.selected_index = 0
|
||||||
|
self._press_time = 0.0
|
||||||
|
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 configdir is None:
|
if configdir is None:
|
||||||
if os.path.isdir("/etc/lxst-client"):
|
if os.path.isdir("/etc/lxst-client"):
|
||||||
self.configdir = "/etc/lxst-client"
|
self.configdir = "/etc/lxst-client"
|
||||||
@@ -99,6 +134,181 @@ class ChannelClient:
|
|||||||
if self.announce_interval > 0:
|
if self.announce_interval > 0:
|
||||||
threading.Thread(target=self._announce_loop, daemon=True).start()
|
threading.Thread(target=self._announce_loop, daemon=True).start()
|
||||||
|
|
||||||
|
# ── Whisplay HAT GUI ─────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _init_whisplay(self):
|
||||||
|
from whisplay_client import create_whisplay_hardware
|
||||||
|
|
||||||
|
self.board = create_whisplay_hardware(
|
||||||
|
app_id="lxst-client",
|
||||||
|
display_name="LXST Client",
|
||||||
|
icon="LC",
|
||||||
|
use_daemon_default_log=True,
|
||||||
|
)
|
||||||
|
self.board.set_backlight(70)
|
||||||
|
self.board.on_button_press(self._on_button_press)
|
||||||
|
self.board.on_button_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._load_fonts()
|
||||||
|
|
||||||
|
def _load_fonts(self):
|
||||||
|
def load(size, bold=False):
|
||||||
|
cand = []
|
||||||
|
if bold:
|
||||||
|
cand.extend([
|
||||||
|
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
|
||||||
|
"/usr/share/fonts/truetype/freefont/FreeSansBold.ttf",
|
||||||
|
])
|
||||||
|
cand.extend([
|
||||||
|
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
||||||
|
"/usr/share/fonts/truetype/freefont/FreeSans.ttf",
|
||||||
|
])
|
||||||
|
for c in cand:
|
||||||
|
if os.path.exists(c):
|
||||||
|
try:
|
||||||
|
return ImageFont.truetype(c, size)
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
return ImageFont.load_default()
|
||||||
|
self._title_font = load(20, True)
|
||||||
|
self._body_font = load(16)
|
||||||
|
self._small_font = load(12)
|
||||||
|
|
||||||
|
def _rgb565(self, image):
|
||||||
|
rgb = image.convert("RGB")
|
||||||
|
out = bytearray()
|
||||||
|
for y in range(rgb.height):
|
||||||
|
for x in range(rgb.width):
|
||||||
|
r, g, b = rgb.getpixel((x, y))
|
||||||
|
v = ((r & 0xF8) << 8) | ((g & 0xFC) << 3) | (b >> 3)
|
||||||
|
out.append((v >> 8) & 0xFF)
|
||||||
|
out.append(v & 0xFF)
|
||||||
|
return bytes(out)
|
||||||
|
|
||||||
|
def _wrap(self, draw, text, font, max_w):
|
||||||
|
if not text:
|
||||||
|
return [""]
|
||||||
|
words = text.split()
|
||||||
|
if not words:
|
||||||
|
return [text]
|
||||||
|
lines, cur = [], words[0]
|
||||||
|
for w in words[1:]:
|
||||||
|
test = f"{cur} {w}"
|
||||||
|
bb = draw.textbbox((0, 0), test, font=font)
|
||||||
|
if (bb[2] - bb[0]) <= max_w:
|
||||||
|
cur = test
|
||||||
|
else:
|
||||||
|
lines.append(cur)
|
||||||
|
cur = w
|
||||||
|
lines.append(cur)
|
||||||
|
return lines
|
||||||
|
|
||||||
|
def _render(self, title, body_lines, accent=(60, 150, 255), footer=""):
|
||||||
|
w, h = self.board.LCD_WIDTH, self.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,
|
||||||
|
fill=(20, 28, 40), outline=(40, 55, 72), width=2)
|
||||||
|
draw.rounded_rectangle((20, 22, w - 20, 54), radius=12, fill=accent)
|
||||||
|
draw.text((28, 30), title, fill=(255, 255, 255), font=self._title_font)
|
||||||
|
y = 70
|
||||||
|
max_w = w - 56
|
||||||
|
for line in body_lines:
|
||||||
|
if y >= h - 50:
|
||||||
|
break
|
||||||
|
if not line:
|
||||||
|
y += 8
|
||||||
|
continue
|
||||||
|
is_sel = False
|
||||||
|
txt = line
|
||||||
|
if isinstance(line, tuple):
|
||||||
|
txt, is_sel = line
|
||||||
|
wrapped = self._wrap(draw, txt, self._body_font, max_w)
|
||||||
|
for wl in wrapped:
|
||||||
|
if is_sel:
|
||||||
|
bb = draw.textbbox((0, 0), wl, font=self._body_font)
|
||||||
|
lw = bb[2] - bb[0]
|
||||||
|
draw.rounded_rectangle((22, y - 2, 22 + lw + 12, y + 20),
|
||||||
|
radius=6, fill=(accent[0] // 3, accent[1] // 3, accent[2] // 3))
|
||||||
|
draw.text((28, y), wl, fill=(255, 255, 255), font=self._body_font)
|
||||||
|
else:
|
||||||
|
draw.text((28, y), wl, fill=(180, 195, 210), font=self._body_font)
|
||||||
|
y += 22
|
||||||
|
if footer:
|
||||||
|
draw.rounded_rectangle((20, h - 38, w - 20, h - 22), radius=10, fill=(28, 37, 50))
|
||||||
|
draw.text((28, h - 34), footer, fill=(140, 195, 255), font=self._small_font)
|
||||||
|
return self._rgb565(img)
|
||||||
|
|
||||||
|
def _flash_screen(self):
|
||||||
|
if self.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)
|
||||||
|
|
||||||
|
def _show_channels(self):
|
||||||
|
if self.board is None:
|
||||||
|
return
|
||||||
|
if not self.channels:
|
||||||
|
fb = self._render("LXST Client", ["No channels", "", "Re-discover to find channels"],
|
||||||
|
accent=(200, 100, 60), footer="No channels")
|
||||||
|
else:
|
||||||
|
lines = []
|
||||||
|
for i, ch in enumerate(self.channels):
|
||||||
|
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)
|
||||||
|
|
||||||
|
def _show_in_call(self):
|
||||||
|
if self.board is None:
|
||||||
|
return
|
||||||
|
ch = self.channels[self.selected_index] if self.channels else None
|
||||||
|
lines = [f"Channel {self.selected_index + 1}"]
|
||||||
|
if ch:
|
||||||
|
lines.append(ch["hash"][:16] + "..")
|
||||||
|
lines += ["", "Press to hang up"]
|
||||||
|
fb = self._render("In Call", lines, accent=(60, 220, 120), footer="Press button to end")
|
||||||
|
self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb)
|
||||||
|
|
||||||
|
def _show_status(self, title, detail="", accent=(255, 180, 60)):
|
||||||
|
if self.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)
|
||||||
|
|
||||||
|
def _on_button_press(self):
|
||||||
|
self._press_time = time.time()
|
||||||
|
|
||||||
|
def _on_button_release(self):
|
||||||
|
elapsed = time.time() - self._press_time
|
||||||
|
if self.state == self.STATE_IN_CALL:
|
||||||
|
self.telephone.hangup()
|
||||||
|
return
|
||||||
|
if self.state != self.STATE_READY:
|
||||||
|
return
|
||||||
|
if elapsed >= 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()
|
||||||
|
|
||||||
|
def _on_exit_request(self, payload=None):
|
||||||
|
self.cleanup()
|
||||||
|
if self.board:
|
||||||
|
self.board.prepare_exit()
|
||||||
|
|
||||||
|
def _on_focus_revoked(self, payload=None):
|
||||||
|
self.cleanup()
|
||||||
|
|
||||||
def _announce_loop(self):
|
def _announce_loop(self):
|
||||||
while self.should_run:
|
while self.should_run:
|
||||||
time.sleep(self.announce_interval)
|
time.sleep(self.announce_interval)
|
||||||
@@ -210,6 +420,9 @@ class ChannelClient:
|
|||||||
f"\nCall established with "
|
f"\nCall established with "
|
||||||
f"{RNS.prettyhexrep(remote_identity.hash)}"
|
f"{RNS.prettyhexrep(remote_identity.hash)}"
|
||||||
)
|
)
|
||||||
|
if self.board:
|
||||||
|
self.board.set_rgb(0, 255, 80)
|
||||||
|
self._show_in_call()
|
||||||
|
|
||||||
def _on_call_ended(self, remote_identity):
|
def _on_call_ended(self, remote_identity):
|
||||||
was_in_call = self.state == self.STATE_IN_CALL
|
was_in_call = self.state == self.STATE_IN_CALL
|
||||||
@@ -219,6 +432,12 @@ class ChannelClient:
|
|||||||
print(f"\nCall ended")
|
print(f"\nCall ended")
|
||||||
elif was_calling:
|
elif was_calling:
|
||||||
print(f"\nCall could not be connected")
|
print(f"\nCall could not be connected")
|
||||||
|
if self.board:
|
||||||
|
self.board.set_rgb(0, 80, 255)
|
||||||
|
if was_in_call:
|
||||||
|
self._show_status("Call Ended")
|
||||||
|
time.sleep(1)
|
||||||
|
self._show_channels()
|
||||||
|
|
||||||
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):
|
||||||
@@ -293,6 +512,26 @@ 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.board:
|
||||||
|
self._run_gui()
|
||||||
|
else:
|
||||||
|
self._run_cli()
|
||||||
|
|
||||||
|
def _run_gui(self):
|
||||||
|
self._flash_screen()
|
||||||
|
self.discover(timeout=15)
|
||||||
|
self.selected_index = 0
|
||||||
|
self.board.set_rgb(0, 80, 255)
|
||||||
|
self._show_channels()
|
||||||
|
while self.should_run:
|
||||||
|
if self.state == self.STATE_CALLING:
|
||||||
|
self._show_status("Calling...", f"Channel {self.selected_index + 1}",
|
||||||
|
accent=(255, 180, 0))
|
||||||
|
time.sleep(0.5)
|
||||||
|
continue
|
||||||
|
time.sleep(0.1)
|
||||||
|
|
||||||
|
def _run_cli(self):
|
||||||
print(
|
print(
|
||||||
f"\n{Terminal.BOLD}"
|
f"\n{Terminal.BOLD}"
|
||||||
f"LXST Channel Client"
|
f"LXST Channel Client"
|
||||||
@@ -359,6 +598,8 @@ 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.board:
|
||||||
|
self.board.cleanup()
|
||||||
|
|
||||||
def _sigint_handler(self, sig, frame):
|
def _sigint_handler(self, sig, frame):
|
||||||
self.cleanup()
|
self.cleanup()
|
||||||
@@ -394,7 +635,11 @@ 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(
|
||||||
@@ -403,6 +648,7 @@ 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:
|
||||||
|
|||||||
@@ -0,0 +1,661 @@
|
|||||||
|
import spidev
|
||||||
|
import time
|
||||||
|
import os
|
||||||
|
import threading
|
||||||
|
import gpiod
|
||||||
|
|
||||||
|
# Detect gpiod API version: v1 has LINE_REQ_DIR_OUT; v2 has LineSettings
|
||||||
|
_GPIOD_V2 = hasattr(gpiod, 'LineSettings')
|
||||||
|
|
||||||
|
if _GPIOD_V2:
|
||||||
|
from gpiod.line import Direction, Value, Bias
|
||||||
|
|
||||||
|
|
||||||
|
# ==================== Platform Detection ====================
|
||||||
|
def _detect_platform():
|
||||||
|
"""Detect hardware platform type"""
|
||||||
|
try:
|
||||||
|
with open("/proc/device-tree/model", "r") as f:
|
||||||
|
model = f.read().strip('\0').strip()
|
||||||
|
if "Raspberry" in model:
|
||||||
|
return "rpi", model
|
||||||
|
elif "Radxa" in model:
|
||||||
|
return "radxa", model
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
# Also check compatible string (some SoCs use non-descriptive model names)
|
||||||
|
try:
|
||||||
|
with open("/proc/device-tree/compatible", "r") as f:
|
||||||
|
compat = f.read()
|
||||||
|
if "radxa" in compat.lower():
|
||||||
|
# Extract first compatible string as readable model name
|
||||||
|
parts = compat.split('\0')
|
||||||
|
model = parts[0] if parts else "Unknown Radxa"
|
||||||
|
return "radxa", model
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return "unknown", "Unknown"
|
||||||
|
|
||||||
|
|
||||||
|
PLATFORM, PLATFORM_MODEL = _detect_platform()
|
||||||
|
|
||||||
|
|
||||||
|
# ==================== Pin Mappings ====================
|
||||||
|
# Physical 40-pin header pin number -> (gpiochip number, line offset)
|
||||||
|
|
||||||
|
# Raspberry Pi: BOARD pin -> BCM GPIO number
|
||||||
|
_RPI_BOARD_TO_BCM = {
|
||||||
|
3: 2, 5: 3, 7: 4, 8: 14,
|
||||||
|
10: 15, 11: 17, 12: 18, 13: 27,
|
||||||
|
15: 22, 16: 23, 18: 24, 19: 10,
|
||||||
|
21: 9, 22: 25, 23: 11, 24: 8,
|
||||||
|
26: 7, 27: 0, 28: 1, 29: 5,
|
||||||
|
31: 6, 32: 12, 33: 13, 35: 19,
|
||||||
|
36: 16, 37: 26, 38: 20, 40: 21,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _build_rpi_pin_map():
|
||||||
|
"""Build RPi BOARD-pin -> (gpiochip, line_offset) mapping.
|
||||||
|
RPi 5 uses gpiochip4 (RP1), earlier models use gpiochip0."""
|
||||||
|
chip_num = 4 if "Raspberry Pi 5" in PLATFORM_MODEL else 0
|
||||||
|
return {pin: (chip_num, bcm) for pin, bcm in _RPI_BOARD_TO_BCM.items()}
|
||||||
|
|
||||||
|
# Based on RK3566 Radxa ZERO 3W
|
||||||
|
RADXA_ZERO3_PIN_MAP = {
|
||||||
|
3: (1, 0), 5: (1, 1), 7: (3, 20), 8: (0, 25),
|
||||||
|
10: (0, 24), 11: (3, 1), 12: (3, 3), 13: (3, 2),
|
||||||
|
15: (3, 8), 16: (3, 9), 18: (3, 10), 19: (4, 19),
|
||||||
|
21: (4, 21), 22: (3, 17), 23: (4, 18), 24: (4, 22),
|
||||||
|
26: (4, 25), 27: (4, 10), 28: (4, 11), 29: (3, 11),
|
||||||
|
31: (3, 12), 32: (3, 18), 33: (3, 19), 35: (3, 4),
|
||||||
|
36: (3, 7), 37: (1, 4), 38: (3, 6), 40: (3, 5),
|
||||||
|
}
|
||||||
|
|
||||||
|
# Based on Allwinner A733 Radxa Cubie A7Z
|
||||||
|
# gpiochip0 (2000000.pinctrl, 352 lines): PA=0-31, PB=32-63, ..., PJ=288-319, PK=320-351
|
||||||
|
# gpiochip1 (7025000.pinctrl, 64 lines): PL=0-31, PM=32-63
|
||||||
|
RADXA_CUBIE_A7Z_PIN_MAP = {
|
||||||
|
3: (0, 311), 5: (0, 310), 7: (0, 32), 8: (0, 41),
|
||||||
|
10: (0, 42), 11: (0, 33), 12: (0, 37), 13: (1, 6),
|
||||||
|
15: (1, 7), 16: (0, 312), 18: (0, 313), 19: (0, 108),
|
||||||
|
21: (0, 109), 22: (1, 5), 23: (0, 107), 24: (0, 106),
|
||||||
|
26: (0, 110), 27: (0, 113), 28: (0, 112), 29: (0, 34),
|
||||||
|
31: (0, 35), 32: (1, 37), 33: (1, 35), 35: (0, 38),
|
||||||
|
36: (0, 36), 37: (1, 36), 38: (0, 40), 40: (0, 39),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _detect_radxa_board():
|
||||||
|
"""Detect specific Radxa board variant from device tree compatible string"""
|
||||||
|
try:
|
||||||
|
with open("/proc/device-tree/compatible", "r") as f:
|
||||||
|
compat = f.read().lower()
|
||||||
|
if "cubie-a7z" in compat:
|
||||||
|
return "cubie-a7z"
|
||||||
|
elif "cubie-a7a" in compat:
|
||||||
|
return "cubie-a7a"
|
||||||
|
elif "cubie-a7s" in compat:
|
||||||
|
return "cubie-a7s"
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
# Default to zero3w for backward compatibility
|
||||||
|
return "zero3w"
|
||||||
|
|
||||||
|
|
||||||
|
# ==================== gpiod v1/v2 Compatibility ====================
|
||||||
|
class _LineHandle:
|
||||||
|
"""Unified thin wrapper that exposes set_value(int) / get_value()->int
|
||||||
|
regardless of whether the underlying library is gpiod v1 or v2."""
|
||||||
|
|
||||||
|
def __init__(self, v2_request=None, v2_offset=None, v1_line=None):
|
||||||
|
self._v2_req = v2_request
|
||||||
|
self._v2_off = v2_offset
|
||||||
|
self._v1 = v1_line
|
||||||
|
|
||||||
|
def set_value(self, val):
|
||||||
|
if self._v2_req is not None:
|
||||||
|
self._v2_req.set_value(
|
||||||
|
self._v2_off, Value.ACTIVE if val else Value.INACTIVE
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._v1.set_value(val)
|
||||||
|
|
||||||
|
def get_value(self):
|
||||||
|
if self._v2_req is not None:
|
||||||
|
return 1 if self._v2_req.get_value(self._v2_off) == Value.ACTIVE else 0
|
||||||
|
return self._v1.get_value()
|
||||||
|
|
||||||
|
def release(self):
|
||||||
|
try:
|
||||||
|
if self._v2_req is not None:
|
||||||
|
self._v2_req.release()
|
||||||
|
else:
|
||||||
|
self._v1.release()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _request_output(chip, line_offset, consumer='whisplay'):
|
||||||
|
"""Request a single GPIO line as output (LOW initial)."""
|
||||||
|
if _GPIOD_V2:
|
||||||
|
settings = gpiod.LineSettings(
|
||||||
|
direction=Direction.OUTPUT, output_value=Value.INACTIVE
|
||||||
|
)
|
||||||
|
req = chip.request_lines(consumer=consumer, config={line_offset: settings})
|
||||||
|
return _LineHandle(v2_request=req, v2_offset=line_offset)
|
||||||
|
else:
|
||||||
|
line = chip.get_line(line_offset)
|
||||||
|
line.request(consumer=consumer, type=gpiod.LINE_REQ_DIR_OUT, default_val=0)
|
||||||
|
return _LineHandle(v1_line=line)
|
||||||
|
|
||||||
|
|
||||||
|
def _request_input(chip, line_offset, consumer='whisplay-btn'):
|
||||||
|
"""Request a single GPIO line as input with bias disabled."""
|
||||||
|
if _GPIOD_V2:
|
||||||
|
try:
|
||||||
|
settings = gpiod.LineSettings(
|
||||||
|
direction=Direction.INPUT, bias=Bias.DISABLED
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
settings = gpiod.LineSettings(direction=Direction.INPUT)
|
||||||
|
req = chip.request_lines(consumer=consumer, config={line_offset: settings})
|
||||||
|
return _LineHandle(v2_request=req, v2_offset=line_offset)
|
||||||
|
else:
|
||||||
|
line = chip.get_line(line_offset)
|
||||||
|
try:
|
||||||
|
line.request(
|
||||||
|
consumer=consumer,
|
||||||
|
type=gpiod.LINE_REQ_DIR_IN,
|
||||||
|
flags=gpiod.LINE_REQ_FLAG_BIAS_DISABLE
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
line.request(consumer=consumer, type=gpiod.LINE_REQ_DIR_IN)
|
||||||
|
return _LineHandle(v1_line=line)
|
||||||
|
|
||||||
|
|
||||||
|
# ==================== Software PWM ====================
|
||||||
|
class SoftPWM:
|
||||||
|
"""Software PWM implementation for GPIO platforms without hardware PWM support"""
|
||||||
|
|
||||||
|
def __init__(self, set_value_func, frequency=100, stop_value=0):
|
||||||
|
self._set_value = set_value_func
|
||||||
|
self.frequency = frequency
|
||||||
|
self.stop_value = stop_value
|
||||||
|
self.duty_cycle = 0.0
|
||||||
|
self._running = False
|
||||||
|
self._thread = None
|
||||||
|
|
||||||
|
def start(self, duty_cycle=0):
|
||||||
|
self.duty_cycle = float(duty_cycle)
|
||||||
|
self._running = True
|
||||||
|
self._thread = threading.Thread(target=self._pwm_loop, daemon=True)
|
||||||
|
self._thread.start()
|
||||||
|
|
||||||
|
def ChangeDutyCycle(self, duty_cycle):
|
||||||
|
self.duty_cycle = max(0.0, min(100.0, float(duty_cycle)))
|
||||||
|
|
||||||
|
def stop(self):
|
||||||
|
self._running = False
|
||||||
|
if self._thread:
|
||||||
|
self._thread.join(timeout=1)
|
||||||
|
try:
|
||||||
|
self._set_value(self.stop_value)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
def _pwm_loop(self):
|
||||||
|
while self._running:
|
||||||
|
period = 1.0 / self.frequency
|
||||||
|
dc = self.duty_cycle
|
||||||
|
if dc <= 0:
|
||||||
|
self._set_value(0)
|
||||||
|
time.sleep(period)
|
||||||
|
elif dc >= 100:
|
||||||
|
self._set_value(1)
|
||||||
|
time.sleep(period)
|
||||||
|
else:
|
||||||
|
on_time = period * dc / 100.0
|
||||||
|
off_time = period - on_time
|
||||||
|
self._set_value(1)
|
||||||
|
time.sleep(on_time)
|
||||||
|
self._set_value(0)
|
||||||
|
time.sleep(off_time)
|
||||||
|
|
||||||
|
|
||||||
|
class WhisplayBoard:
|
||||||
|
# LCD parameters
|
||||||
|
LCD_WIDTH = 240
|
||||||
|
LCD_HEIGHT = 280
|
||||||
|
BUTTON_POLL_INTERVAL_SEC = 0.005
|
||||||
|
CornerHeight = 20 # Rounded corner height in pixels
|
||||||
|
|
||||||
|
# Physical pin definitions (BOARD mode - shared by both platforms)
|
||||||
|
DC_PIN = 13
|
||||||
|
RST_PIN = 7
|
||||||
|
LED_PIN = 15
|
||||||
|
|
||||||
|
# RGB LED pins
|
||||||
|
RED_PIN = 22
|
||||||
|
GREEN_PIN = 18
|
||||||
|
BLUE_PIN = 16
|
||||||
|
|
||||||
|
# Button pin
|
||||||
|
BUTTON_PIN = 11
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.platform = PLATFORM
|
||||||
|
self.backlight_pwm = None
|
||||||
|
self._current_r = 0
|
||||||
|
self._current_g = 0
|
||||||
|
self._current_b = 0
|
||||||
|
self.button_press_callback = None
|
||||||
|
self.button_release_callback = None
|
||||||
|
|
||||||
|
# Select pin map and SPI config based on platform
|
||||||
|
if self.platform == "rpi":
|
||||||
|
self._pin_map = _build_rpi_pin_map()
|
||||||
|
self._spi_bus = 0
|
||||||
|
self._spi_cs = 0
|
||||||
|
self._spi_speed = 100_000_000
|
||||||
|
elif self.platform == "radxa":
|
||||||
|
self._radxa_board = _detect_radxa_board()
|
||||||
|
if self._radxa_board == "cubie-a7z":
|
||||||
|
self._pin_map = RADXA_CUBIE_A7Z_PIN_MAP
|
||||||
|
self._spi_bus = 1 # SPI1, CS0 (Allwinner A733)
|
||||||
|
self._spi_cs = 0
|
||||||
|
self._spi_speed = 48_000_000
|
||||||
|
else:
|
||||||
|
self._pin_map = RADXA_ZERO3_PIN_MAP
|
||||||
|
self._spi_bus = 3 # SPI3, CS0 (RK3566 Radxa Zero 3W)
|
||||||
|
self._spi_cs = 0
|
||||||
|
self._spi_speed = 48_000_000
|
||||||
|
else:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Unsupported platform: {self.platform}\n"
|
||||||
|
"Install python3-libgpiod: sudo apt install python3-libgpiod"
|
||||||
|
)
|
||||||
|
|
||||||
|
self._init_gpio()
|
||||||
|
self._init_spi()
|
||||||
|
|
||||||
|
self.previous_frame = None
|
||||||
|
# Detect hardware version and set backlight mode
|
||||||
|
self._detect_hardware_version()
|
||||||
|
self._detect_wm8960()
|
||||||
|
self.set_backlight(0)
|
||||||
|
self._reset_lcd()
|
||||||
|
self._init_display()
|
||||||
|
self.fill_screen(0)
|
||||||
|
|
||||||
|
# ==================== GPIO Initialization (gpiod, unified) ====================
|
||||||
|
def _init_gpio(self):
|
||||||
|
"""Initialize all GPIO lines via gpiod (works on both RPi and Radxa)."""
|
||||||
|
self._gpio_chips = {}
|
||||||
|
self._gpio_lines = {}
|
||||||
|
|
||||||
|
pins_used = [self.DC_PIN, self.RST_PIN, self.LED_PIN,
|
||||||
|
self.RED_PIN, self.GREEN_PIN, self.BLUE_PIN,
|
||||||
|
self.BUTTON_PIN]
|
||||||
|
|
||||||
|
for pin in pins_used:
|
||||||
|
if pin not in self._pin_map:
|
||||||
|
raise RuntimeError(f"Physical pin {pin} is not defined in pin map")
|
||||||
|
chip_num, _ = self._pin_map[pin]
|
||||||
|
if chip_num not in self._gpio_chips:
|
||||||
|
self._gpio_chips[chip_num] = gpiod.Chip(f'/dev/gpiochip{chip_num}')
|
||||||
|
|
||||||
|
# Request output pins
|
||||||
|
output_pins = [self.DC_PIN, self.RST_PIN, self.LED_PIN,
|
||||||
|
self.RED_PIN, self.GREEN_PIN, self.BLUE_PIN]
|
||||||
|
for pin in output_pins:
|
||||||
|
chip_num, line_offset = self._pin_map[pin]
|
||||||
|
chip = self._gpio_chips[chip_num]
|
||||||
|
self._gpio_lines[pin] = _request_output(chip, line_offset)
|
||||||
|
|
||||||
|
# Enable backlight (LOW = on)
|
||||||
|
self._gpio_lines[self.LED_PIN].set_value(0)
|
||||||
|
|
||||||
|
# Initialize RGB LED (using software PWM)
|
||||||
|
red_line = self._gpio_lines[self.RED_PIN]
|
||||||
|
green_line = self._gpio_lines[self.GREEN_PIN]
|
||||||
|
blue_line = self._gpio_lines[self.BLUE_PIN]
|
||||||
|
self.red_pwm = SoftPWM(red_line.set_value, 100, stop_value=1)
|
||||||
|
self.green_pwm = SoftPWM(green_line.set_value, 100, stop_value=1)
|
||||||
|
self.blue_pwm = SoftPWM(blue_line.set_value, 100, stop_value=1)
|
||||||
|
self.red_pwm.start(0)
|
||||||
|
self.green_pwm.start(0)
|
||||||
|
self.blue_pwm.start(0)
|
||||||
|
|
||||||
|
# Initialize button (input, polled for state changes)
|
||||||
|
# The WhisPlay HAT has an external pull-down resistor on the button line.
|
||||||
|
# Button pressed = HIGH, released = LOW. No internal pull needed.
|
||||||
|
chip_num, line_offset = self._pin_map[self.BUTTON_PIN]
|
||||||
|
chip = self._gpio_chips[chip_num]
|
||||||
|
self._gpio_lines[self.BUTTON_PIN] = _request_input(chip, line_offset)
|
||||||
|
|
||||||
|
# Start button event listener thread (polling)
|
||||||
|
self._btn_thread_running = True
|
||||||
|
self._btn_thread = threading.Thread(target=self._button_monitor, daemon=True)
|
||||||
|
self._btn_thread.start()
|
||||||
|
|
||||||
|
def _init_spi(self):
|
||||||
|
"""Initialize SPI bus."""
|
||||||
|
self.spi = spidev.SpiDev()
|
||||||
|
self.spi.open(self._spi_bus, self._spi_cs)
|
||||||
|
self.spi.max_speed_hz = self._spi_speed
|
||||||
|
self.spi.mode = 0b00
|
||||||
|
|
||||||
|
def _button_monitor(self):
|
||||||
|
"""Button state polling thread.
|
||||||
|
HIGH (1) = pressed, LOW (0) = released.
|
||||||
|
Poll interval is configurable; shorter values improve responsiveness.
|
||||||
|
"""
|
||||||
|
btn_line = self._gpio_lines[self.BUTTON_PIN]
|
||||||
|
last_state = btn_line.get_value()
|
||||||
|
poll_interval = self.BUTTON_POLL_INTERVAL_SEC
|
||||||
|
while self._btn_thread_running:
|
||||||
|
try:
|
||||||
|
state = btn_line.get_value()
|
||||||
|
if state != last_state:
|
||||||
|
last_state = state
|
||||||
|
if state == 1:
|
||||||
|
if self.button_press_callback:
|
||||||
|
self.button_press_callback()
|
||||||
|
else:
|
||||||
|
if self.button_release_callback:
|
||||||
|
self.button_release_callback()
|
||||||
|
except Exception:
|
||||||
|
if self._btn_thread_running:
|
||||||
|
pass
|
||||||
|
time.sleep(poll_interval)
|
||||||
|
|
||||||
|
# ==================== GPIO Helpers ====================
|
||||||
|
def _gpio_output(self, pin, value):
|
||||||
|
"""Set GPIO pin output value"""
|
||||||
|
self._gpio_lines[pin].set_value(1 if value else 0)
|
||||||
|
|
||||||
|
def _gpio_input(self, pin):
|
||||||
|
"""Read GPIO pin input value"""
|
||||||
|
return self._gpio_lines[pin].get_value()
|
||||||
|
|
||||||
|
# ==================== Hardware Detection ====================
|
||||||
|
def _detect_hardware_version(self):
|
||||||
|
"""Detect hardware version and set backlight mode accordingly"""
|
||||||
|
try:
|
||||||
|
model = PLATFORM_MODEL
|
||||||
|
if self.platform == "rpi":
|
||||||
|
if "Zero" in model and "2" not in model:
|
||||||
|
self.backlight_mode = False # Use simple on/off mode
|
||||||
|
else:
|
||||||
|
self.backlight_mode = True # Use PWM mode
|
||||||
|
elif self.platform == "radxa":
|
||||||
|
# Radxa uses software PWM mode
|
||||||
|
self.backlight_mode = True
|
||||||
|
else:
|
||||||
|
self.backlight_mode = True
|
||||||
|
print(
|
||||||
|
f"Detected hardware: {model}, Backlight mode: {'PWM' if self.backlight_mode else 'Simple Switch'}")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Error detecting hardware version: {e}")
|
||||||
|
self.backlight_mode = True
|
||||||
|
|
||||||
|
def _detect_wm8960(self):
|
||||||
|
"""Detect if a sound card containing wm8960 exists"""
|
||||||
|
try:
|
||||||
|
with open("/proc/asound/cards", "r") as f:
|
||||||
|
lines = f.readlines()
|
||||||
|
for line in lines:
|
||||||
|
if "wm8960" in line.lower():
|
||||||
|
print("wm8960 sound card detected.")
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Error detecting wm8960 sound card: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
print("wm8960 sound card not detected. Please refer to the following page for installation instructions.")
|
||||||
|
print("https://docs.pisugar.com/")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# ========== Backlight Control ==========
|
||||||
|
def set_backlight(self, brightness):
|
||||||
|
if self.backlight_mode: # PWM mode
|
||||||
|
if self.backlight_pwm is None:
|
||||||
|
led_line = self._gpio_lines[self.LED_PIN]
|
||||||
|
self.backlight_pwm = SoftPWM(led_line.set_value, 1000, stop_value=1)
|
||||||
|
self.backlight_pwm.start(100)
|
||||||
|
if 0 <= brightness <= 100:
|
||||||
|
duty_cycle = 100 - brightness
|
||||||
|
self.backlight_pwm.ChangeDutyCycle(duty_cycle)
|
||||||
|
else: # Simple on/off mode
|
||||||
|
if brightness == 0:
|
||||||
|
self._gpio_output(self.LED_PIN, 1) # Turn off backlight
|
||||||
|
else:
|
||||||
|
self._gpio_output(self.LED_PIN, 0) # Turn on backlight
|
||||||
|
|
||||||
|
def set_backlight_mode(self, mode):
|
||||||
|
"""
|
||||||
|
Set backlight mode
|
||||||
|
:param mode: True for PWM brightness control, False for simple on/off
|
||||||
|
"""
|
||||||
|
if mode == self.backlight_mode:
|
||||||
|
return # Mode unchanged, no action needed
|
||||||
|
|
||||||
|
if mode: # Switch to PWM mode
|
||||||
|
led_line = self._gpio_lines[self.LED_PIN]
|
||||||
|
self.backlight_pwm = SoftPWM(led_line.set_value, 1000, stop_value=1)
|
||||||
|
self.backlight_pwm.start(100)
|
||||||
|
else: # Switch to simple on/off mode
|
||||||
|
if self.backlight_pwm is not None:
|
||||||
|
self.backlight_pwm.stop()
|
||||||
|
self.backlight_pwm = None
|
||||||
|
self._gpio_output(self.LED_PIN, 1) # Ensure backlight is on
|
||||||
|
self.backlight_mode = mode
|
||||||
|
|
||||||
|
def _reset_lcd(self):
|
||||||
|
self._gpio_output(self.RST_PIN, 1)
|
||||||
|
time.sleep(0.1)
|
||||||
|
self._gpio_output(self.RST_PIN, 0)
|
||||||
|
time.sleep(0.1)
|
||||||
|
self._gpio_output(self.RST_PIN, 1)
|
||||||
|
time.sleep(0.12)
|
||||||
|
|
||||||
|
def _init_display(self):
|
||||||
|
self._send_command(0x11)
|
||||||
|
time.sleep(0.12)
|
||||||
|
USE_HORIZONTAL = 1
|
||||||
|
direction = {0: 0x00, 1: 0xC0, 2: 0x70,
|
||||||
|
3: 0xA0}.get(USE_HORIZONTAL, 0x00)
|
||||||
|
self._send_command(0x36, direction)
|
||||||
|
self._send_command(0x3A, 0x05)
|
||||||
|
self._send_command(0xB2, 0x0C, 0x0C, 0x00, 0x33, 0x33)
|
||||||
|
self._send_command(0xB7, 0x35)
|
||||||
|
self._send_command(0xBB, 0x32)
|
||||||
|
self._send_command(0xC2, 0x01)
|
||||||
|
self._send_command(0xC3, 0x15)
|
||||||
|
self._send_command(0xC4, 0x20)
|
||||||
|
self._send_command(0xC6, 0x0F)
|
||||||
|
self._send_command(0xD0, 0xA4, 0xA1)
|
||||||
|
self._send_command(
|
||||||
|
0xE0,
|
||||||
|
0xD0,
|
||||||
|
0x08,
|
||||||
|
0x0E,
|
||||||
|
0x09,
|
||||||
|
0x09,
|
||||||
|
0x05,
|
||||||
|
0x31,
|
||||||
|
0x33,
|
||||||
|
0x48,
|
||||||
|
0x17,
|
||||||
|
0x14,
|
||||||
|
0x15,
|
||||||
|
0x31,
|
||||||
|
0x34,
|
||||||
|
)
|
||||||
|
self._send_command(
|
||||||
|
0xE1,
|
||||||
|
0xD0,
|
||||||
|
0x08,
|
||||||
|
0x0E,
|
||||||
|
0x09,
|
||||||
|
0x09,
|
||||||
|
0x15,
|
||||||
|
0x31,
|
||||||
|
0x33,
|
||||||
|
0x48,
|
||||||
|
0x17,
|
||||||
|
0x14,
|
||||||
|
0x15,
|
||||||
|
0x31,
|
||||||
|
0x34,
|
||||||
|
)
|
||||||
|
self._send_command(0x21)
|
||||||
|
self._send_command(0x29)
|
||||||
|
|
||||||
|
def _send_command(self, cmd, *args):
|
||||||
|
self._gpio_output(self.DC_PIN, 0)
|
||||||
|
self.spi.xfer2([cmd])
|
||||||
|
if args:
|
||||||
|
self._gpio_output(self.DC_PIN, 1)
|
||||||
|
self._send_data(list(args))
|
||||||
|
|
||||||
|
def _send_data(self, data):
|
||||||
|
self._gpio_output(self.DC_PIN, 1)
|
||||||
|
try:
|
||||||
|
self.spi.writebytes2(data)
|
||||||
|
except AttributeError:
|
||||||
|
max_chunk = 4096
|
||||||
|
for i in range(0, len(data), max_chunk):
|
||||||
|
self.spi.writebytes(data[i : i + max_chunk])
|
||||||
|
|
||||||
|
def _send_data_bytes(self, data: bytes | bytearray):
|
||||||
|
"""Fast path for bytes/bytearray — avoids Python list overhead."""
|
||||||
|
self._gpio_output(self.DC_PIN, 1)
|
||||||
|
try:
|
||||||
|
self.spi.writebytes2(data)
|
||||||
|
except AttributeError:
|
||||||
|
max_chunk = 4096
|
||||||
|
for i in range(0, len(data), max_chunk):
|
||||||
|
self.spi.writebytes(list(data[i : i + max_chunk]))
|
||||||
|
|
||||||
|
def set_window(self, x0, y0, x1, y1, use_horizontal=0):
|
||||||
|
if use_horizontal in (0, 1):
|
||||||
|
self._send_command(0x2A, x0 >> 8, x0 & 0xFF, x1 >> 8, x1 & 0xFF)
|
||||||
|
self._send_command(
|
||||||
|
0x2B, (y0 + 20) >> 8, (y0 + 20) & 0xFF, (y1 +
|
||||||
|
20) >> 8, (y1 + 20) & 0xFF
|
||||||
|
)
|
||||||
|
elif use_horizontal in (2, 3):
|
||||||
|
self._send_command(
|
||||||
|
0x2A, (x0 + 20) >> 8, (x0 + 20) & 0xFF, (x1 +
|
||||||
|
20) >> 8, (x1 + 20) & 0xFF
|
||||||
|
)
|
||||||
|
self._send_command(0x2B, y0 >> 8, y0 & 0xFF, y1 >> 8, y1 & 0xFF)
|
||||||
|
self._send_command(0x2C)
|
||||||
|
|
||||||
|
def draw_pixel(self, x, y, color):
|
||||||
|
if x >= self.LCD_WIDTH or y >= self.LCD_HEIGHT:
|
||||||
|
return
|
||||||
|
self.set_window(x, y, x, y)
|
||||||
|
self._send_data([(color >> 8) & 0xFF, color & 0xFF])
|
||||||
|
|
||||||
|
def draw_line(self, x0, y0, x1, y1, color):
|
||||||
|
dx = abs(x1 - x0)
|
||||||
|
dy = abs(y1 - y0)
|
||||||
|
sx = 1 if x0 < x1 else -1
|
||||||
|
sy = 1 if y0 < y1 else -1
|
||||||
|
err = dx - dy
|
||||||
|
|
||||||
|
while True:
|
||||||
|
self.draw_pixel(x0, y0, color)
|
||||||
|
if x0 == x1 and y0 == y1:
|
||||||
|
break
|
||||||
|
e2 = 2 * err
|
||||||
|
if e2 > -dy:
|
||||||
|
err -= dy
|
||||||
|
x0 += sx
|
||||||
|
if e2 < dx:
|
||||||
|
err += dx
|
||||||
|
y0 += sy
|
||||||
|
|
||||||
|
def fill_screen(self, color):
|
||||||
|
self.set_window(0, 0, self.LCD_WIDTH - 1, self.LCD_HEIGHT - 1)
|
||||||
|
high = (color >> 8) & 0xFF
|
||||||
|
low = color & 0xFF
|
||||||
|
buffer = bytes([high, low]) * (self.LCD_WIDTH * self.LCD_HEIGHT)
|
||||||
|
self._send_data_bytes(buffer)
|
||||||
|
|
||||||
|
def draw_image(self, x, y, width, height, pixel_data):
|
||||||
|
if (x + width > self.LCD_WIDTH) or (y + height > self.LCD_HEIGHT):
|
||||||
|
raise ValueError("Image dimensions exceed screen bounds")
|
||||||
|
self.set_window(x, y, x + width - 1, y + height - 1)
|
||||||
|
if isinstance(pixel_data, (bytes, bytearray)):
|
||||||
|
self._send_data_bytes(pixel_data)
|
||||||
|
else:
|
||||||
|
self._send_data(pixel_data)
|
||||||
|
|
||||||
|
# ========== RGB LED & Button ==========
|
||||||
|
def set_rgb(self, r, g, b):
|
||||||
|
self.red_pwm.ChangeDutyCycle(100 - (r / 255 * 100))
|
||||||
|
self.green_pwm.ChangeDutyCycle(100 - (g / 255 * 100))
|
||||||
|
self.blue_pwm.ChangeDutyCycle(100 - (b / 255 * 100))
|
||||||
|
self._current_r = r
|
||||||
|
self._current_g = g
|
||||||
|
self._current_b = b
|
||||||
|
|
||||||
|
def set_rgb_fade(self, r_target, g_target, b_target, duration_ms=100):
|
||||||
|
steps = 20 # Adjust steps to control fade smoothness
|
||||||
|
delay_ms = duration_ms / steps
|
||||||
|
|
||||||
|
r_step = (r_target - self._current_r) / steps
|
||||||
|
g_step = (g_target - self._current_g) / steps
|
||||||
|
b_step = (b_target - self._current_b) / steps
|
||||||
|
|
||||||
|
for _ in range(steps + 1):
|
||||||
|
r_interim = int(self._current_r + _ * r_step)
|
||||||
|
g_interim = int(self._current_g + _ * g_step)
|
||||||
|
b_interim = int(self._current_b + _ * b_step)
|
||||||
|
self.set_rgb(
|
||||||
|
max(0, min(255, r_interim)),
|
||||||
|
max(0, min(255, g_interim)),
|
||||||
|
max(0, min(255, b_interim)),
|
||||||
|
)
|
||||||
|
time.sleep(delay_ms / 1000.0)
|
||||||
|
|
||||||
|
def button_pressed(self):
|
||||||
|
return self._gpio_input(self.BUTTON_PIN) == 1
|
||||||
|
|
||||||
|
def on_button_press(self, callback):
|
||||||
|
self.button_press_callback = callback
|
||||||
|
|
||||||
|
def on_button_release(self, callback):
|
||||||
|
self.button_release_callback = callback
|
||||||
|
|
||||||
|
# ========== Cleanup ==========
|
||||||
|
def cleanup(self):
|
||||||
|
# Stop backlight PWM
|
||||||
|
if self.backlight_pwm is not None:
|
||||||
|
self.backlight_pwm.stop()
|
||||||
|
# Close SPI
|
||||||
|
self.spi.close()
|
||||||
|
# Stop RGB LED PWM
|
||||||
|
self.red_pwm.stop()
|
||||||
|
self.green_pwm.stop()
|
||||||
|
self.blue_pwm.stop()
|
||||||
|
|
||||||
|
# Stop button listener thread
|
||||||
|
self._btn_thread_running = False
|
||||||
|
if hasattr(self, '_btn_thread') and self._btn_thread:
|
||||||
|
self._btn_thread.join(timeout=2)
|
||||||
|
# Release GPIO resources
|
||||||
|
for line in self._gpio_lines.values():
|
||||||
|
try:
|
||||||
|
line.release()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
for chip in self._gpio_chips.values():
|
||||||
|
try:
|
||||||
|
chip.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
@@ -0,0 +1,281 @@
|
|||||||
|
import json
|
||||||
|
import mmap
|
||||||
|
import socket
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
from whisplay import WhisplayBoard
|
||||||
|
|
||||||
|
|
||||||
|
DEFAULT_DAEMON_SOCKET_PATH = "/tmp/whisplay-daemon.sock"
|
||||||
|
DEFAULT_APP_ID = "whisplay-ai-chatbot"
|
||||||
|
DEFAULT_APP_DISPLAY_NAME = "AI Chatbot"
|
||||||
|
DEFAULT_APP_ICON = "AI"
|
||||||
|
DEFAULT_EXIT_GESTURE = "quad_click"
|
||||||
|
DEFAULT_PRIORITY = 0
|
||||||
|
DEFAULT_USE_DAEMON_DEFAULT_LOG = False
|
||||||
|
|
||||||
|
|
||||||
|
class WhisplayDaemonProxy:
|
||||||
|
LCD_WIDTH = 240
|
||||||
|
LCD_HEIGHT = 280
|
||||||
|
CornerHeight = 20
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
socket_path: str = DEFAULT_DAEMON_SOCKET_PATH,
|
||||||
|
app_id: str = DEFAULT_APP_ID,
|
||||||
|
display_name: str = DEFAULT_APP_DISPLAY_NAME,
|
||||||
|
icon: str = DEFAULT_APP_ICON,
|
||||||
|
launch_command: str | None = None,
|
||||||
|
launch_cwd: str | None = None,
|
||||||
|
persist: bool = True,
|
||||||
|
exit_gesture: str = DEFAULT_EXIT_GESTURE,
|
||||||
|
priority: int = DEFAULT_PRIORITY,
|
||||||
|
use_daemon_default_log: bool = DEFAULT_USE_DAEMON_DEFAULT_LOG,
|
||||||
|
):
|
||||||
|
self.socket_path = socket_path
|
||||||
|
self.button_press_callback = None
|
||||||
|
self.button_release_callback = None
|
||||||
|
self.exit_request_callback = None
|
||||||
|
self.focus_revoked_callback = None
|
||||||
|
self._button_down = False
|
||||||
|
self._subscriber = None
|
||||||
|
self._running = False
|
||||||
|
self._mmap = None
|
||||||
|
self._fb_file = None
|
||||||
|
self._fb_stride = self.LCD_WIDTH * 2
|
||||||
|
self._fb_path = None
|
||||||
|
self._session_token = None
|
||||||
|
self._app_id = app_id
|
||||||
|
self._display_name = display_name
|
||||||
|
self._icon = icon
|
||||||
|
self._launch_command = launch_command
|
||||||
|
self._launch_cwd = launch_cwd
|
||||||
|
self._persist = persist
|
||||||
|
self._exit_gesture = str(exit_gesture or DEFAULT_EXIT_GESTURE)
|
||||||
|
self._priority = int(priority)
|
||||||
|
self._use_daemon_default_log = bool(use_daemon_default_log)
|
||||||
|
|
||||||
|
def _send_request(self, cmd: str, payload: dict | None = None) -> dict:
|
||||||
|
body = {"version": 1, "cmd": cmd, "payload": payload or {}}
|
||||||
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
|
||||||
|
client.connect(self.socket_path)
|
||||||
|
client.sendall((json.dumps(body) + "\n").encode("utf-8"))
|
||||||
|
line = client.makefile("r").readline().strip()
|
||||||
|
if not line:
|
||||||
|
raise RuntimeError("empty response from whisplay-daemon")
|
||||||
|
response = json.loads(line)
|
||||||
|
if not response.get("ok"):
|
||||||
|
raise RuntimeError(response.get("error", "whisplay-daemon request failed"))
|
||||||
|
return response
|
||||||
|
|
||||||
|
def ping(self) -> bool:
|
||||||
|
try:
|
||||||
|
self._send_request("health.ping")
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def register(self):
|
||||||
|
payload = {
|
||||||
|
"app_id": self._app_id,
|
||||||
|
"display_name": self._display_name,
|
||||||
|
"icon": self._icon,
|
||||||
|
"persist": self._persist,
|
||||||
|
}
|
||||||
|
if self._launch_command is not None:
|
||||||
|
payload["launch_command"] = self._launch_command
|
||||||
|
if self._launch_cwd is not None:
|
||||||
|
payload["cwd"] = self._launch_cwd
|
||||||
|
payload["exit_gesture"] = self._exit_gesture
|
||||||
|
payload["priority"] = self._priority
|
||||||
|
payload["use_daemon_default_log"] = self._use_daemon_default_log
|
||||||
|
self._send_request("app.register", payload)
|
||||||
|
|
||||||
|
def acquire_foreground(self, timeout_sec: float = 5.0):
|
||||||
|
deadline = time.time() + timeout_sec
|
||||||
|
last_error = None
|
||||||
|
while time.time() < deadline:
|
||||||
|
try:
|
||||||
|
response = self._send_request("app.focus.acquire", {"app_id": self._app_id})
|
||||||
|
self._session_token = response["payload"]["session_token"]
|
||||||
|
fb = self._send_request(
|
||||||
|
"framebuffer.acquire",
|
||||||
|
{"app_id": self._app_id, "session_token": self._session_token},
|
||||||
|
)["payload"]
|
||||||
|
self._attach_framebuffer(fb["buffer_handle"], int(fb["stride"]))
|
||||||
|
return
|
||||||
|
except Exception as exc:
|
||||||
|
last_error = exc
|
||||||
|
time.sleep(0.2)
|
||||||
|
raise RuntimeError(f"failed to acquire foreground: {last_error}")
|
||||||
|
|
||||||
|
def _attach_framebuffer(self, buffer_handle: str, stride: int):
|
||||||
|
self._detach_framebuffer()
|
||||||
|
self._fb_path = buffer_handle
|
||||||
|
self._fb_stride = stride
|
||||||
|
self._fb_file = open(buffer_handle, "r+b")
|
||||||
|
self._mmap = mmap.mmap(self._fb_file.fileno(), 0)
|
||||||
|
|
||||||
|
def _detach_framebuffer(self):
|
||||||
|
if self._mmap is not None:
|
||||||
|
try:
|
||||||
|
self._mmap.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self._mmap = None
|
||||||
|
if self._fb_file is not None:
|
||||||
|
try:
|
||||||
|
self._fb_file.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self._fb_file = None
|
||||||
|
self._fb_path = None
|
||||||
|
|
||||||
|
def release_focus(self):
|
||||||
|
if self._session_token:
|
||||||
|
try:
|
||||||
|
self._send_request(
|
||||||
|
"app.focus.release",
|
||||||
|
{"app_id": self._app_id, "session_token": self._session_token},
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self._session_token = None
|
||||||
|
self._detach_framebuffer()
|
||||||
|
|
||||||
|
def prepare_exit(self):
|
||||||
|
self.release_focus()
|
||||||
|
|
||||||
|
def set_backlight(self, brightness):
|
||||||
|
self._send_request("backlight.set", {"brightness": int(brightness)})
|
||||||
|
|
||||||
|
def set_rgb(self, r, g, b):
|
||||||
|
self._send_request("led.set", {"r": int(r), "g": int(g), "b": int(b)})
|
||||||
|
|
||||||
|
def set_rgb_fade(self, r_target, g_target, b_target, duration_ms=100):
|
||||||
|
self._send_request(
|
||||||
|
"led.fade",
|
||||||
|
{
|
||||||
|
"r": int(r_target),
|
||||||
|
"g": int(g_target),
|
||||||
|
"b": int(b_target),
|
||||||
|
"duration_ms": int(duration_ms),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
def draw_image(self, x, y, width, height, pixel_data):
|
||||||
|
if self._mmap is None:
|
||||||
|
return
|
||||||
|
frame_bytes = bytes(pixel_data if not isinstance(pixel_data, bytes) else pixel_data)
|
||||||
|
row_bytes = width * 2
|
||||||
|
for row in range(height):
|
||||||
|
src = row * row_bytes
|
||||||
|
dst = ((y + row) * self._fb_stride) + (x * 2)
|
||||||
|
self._mmap[dst:dst + row_bytes] = frame_bytes[src:src + row_bytes]
|
||||||
|
|
||||||
|
def fill_screen(self, color):
|
||||||
|
if self._mmap is None:
|
||||||
|
return
|
||||||
|
high = (int(color) >> 8) & 0xFF
|
||||||
|
low = int(color) & 0xFF
|
||||||
|
self._mmap.seek(0)
|
||||||
|
self._mmap.write(bytes([high, low]) * (self.LCD_WIDTH * self.LCD_HEIGHT))
|
||||||
|
self._mmap.seek(0)
|
||||||
|
|
||||||
|
def button_pressed(self):
|
||||||
|
return self._button_down
|
||||||
|
|
||||||
|
def on_button_press(self, callback):
|
||||||
|
self.button_press_callback = callback
|
||||||
|
|
||||||
|
def on_button_release(self, callback):
|
||||||
|
self.button_release_callback = callback
|
||||||
|
|
||||||
|
def on_exit_request(self, callback):
|
||||||
|
self.exit_request_callback = callback
|
||||||
|
|
||||||
|
def on_focus_revoked(self, callback):
|
||||||
|
self.focus_revoked_callback = callback
|
||||||
|
|
||||||
|
def start_event_listener(self):
|
||||||
|
if self._subscriber is not None:
|
||||||
|
return
|
||||||
|
self._running = True
|
||||||
|
self._subscriber = threading.Thread(target=self._event_loop, daemon=True)
|
||||||
|
self._subscriber.start()
|
||||||
|
|
||||||
|
def _event_loop(self):
|
||||||
|
while self._running:
|
||||||
|
try:
|
||||||
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
|
||||||
|
client.connect(self.socket_path)
|
||||||
|
body = {"version": 1, "cmd": "events.subscribe", "payload": {"app_id": self._app_id}}
|
||||||
|
client.sendall((json.dumps(body) + "\n").encode("utf-8"))
|
||||||
|
reader = client.makefile("r")
|
||||||
|
ack = reader.readline().strip()
|
||||||
|
if not ack:
|
||||||
|
raise RuntimeError("subscription ack missing")
|
||||||
|
for line in reader:
|
||||||
|
if not self._running:
|
||||||
|
return
|
||||||
|
line = line.strip()
|
||||||
|
if not line:
|
||||||
|
continue
|
||||||
|
event = json.loads(line)
|
||||||
|
name = event.get("event")
|
||||||
|
payload = event.get("payload", {}) or {}
|
||||||
|
if name == "button_pressed":
|
||||||
|
self._button_down = True
|
||||||
|
if self.button_press_callback:
|
||||||
|
self.button_press_callback()
|
||||||
|
elif name == "button_released":
|
||||||
|
self._button_down = False
|
||||||
|
if self.button_release_callback:
|
||||||
|
self.button_release_callback()
|
||||||
|
elif name == "app_exit_requested":
|
||||||
|
if self.exit_request_callback:
|
||||||
|
self.exit_request_callback()
|
||||||
|
elif name == "app_focus_revoked":
|
||||||
|
self._session_token = None
|
||||||
|
self._detach_framebuffer()
|
||||||
|
if self.focus_revoked_callback:
|
||||||
|
self.focus_revoked_callback(payload)
|
||||||
|
except Exception:
|
||||||
|
time.sleep(0.5)
|
||||||
|
|
||||||
|
def cleanup(self):
|
||||||
|
self._running = False
|
||||||
|
self.release_focus()
|
||||||
|
|
||||||
|
|
||||||
|
def create_whisplay_hardware(
|
||||||
|
app_id: str = DEFAULT_APP_ID,
|
||||||
|
display_name: str = DEFAULT_APP_DISPLAY_NAME,
|
||||||
|
icon: str = DEFAULT_APP_ICON,
|
||||||
|
launch_command: str | None = None,
|
||||||
|
launch_cwd: str | None = None,
|
||||||
|
persist: bool = True,
|
||||||
|
exit_gesture: str = DEFAULT_EXIT_GESTURE,
|
||||||
|
priority: int = DEFAULT_PRIORITY,
|
||||||
|
use_daemon_default_log: bool = DEFAULT_USE_DAEMON_DEFAULT_LOG,
|
||||||
|
):
|
||||||
|
daemon = WhisplayDaemonProxy(
|
||||||
|
socket_path=DEFAULT_DAEMON_SOCKET_PATH,
|
||||||
|
app_id=app_id,
|
||||||
|
display_name=display_name,
|
||||||
|
icon=icon,
|
||||||
|
launch_command=launch_command,
|
||||||
|
launch_cwd=launch_cwd,
|
||||||
|
persist=persist,
|
||||||
|
exit_gesture=exit_gesture,
|
||||||
|
priority=priority,
|
||||||
|
use_daemon_default_log=use_daemon_default_log,
|
||||||
|
)
|
||||||
|
if daemon.ping():
|
||||||
|
daemon.register()
|
||||||
|
daemon.start_event_listener()
|
||||||
|
daemon.acquire_foreground()
|
||||||
|
return daemon
|
||||||
|
return WhisplayBoard()
|
||||||
Reference in New Issue
Block a user