forked from iarv/contact
Compare commits
30 Commits
1.4.2
...
close-inte
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6b18809215 | ||
|
|
b048fe2480 | ||
|
|
600fc61ed7 | ||
|
|
fbf5ff6bd3 | ||
|
|
faab1e961f | ||
|
|
255db3aa3c | ||
|
|
42717c956f | ||
|
|
ad77eba0d6 | ||
|
|
7d6918c69e | ||
|
|
70646a1214 | ||
|
|
53c1320d87 | ||
|
|
ed9ff60f97 | ||
|
|
443df7bf48 | ||
|
|
d8452e74d5 | ||
|
|
2cefdfb645 | ||
|
|
191d6bad35 | ||
|
|
bf1d0ecea9 | ||
|
|
33904d2785 | ||
|
|
b5fd8d74c4 | ||
|
|
c383091a00 | ||
|
|
cc37f9a66b | ||
|
|
41ea441e32 | ||
|
|
58fb82fb1b | ||
|
|
dcd39c231f | ||
|
|
87bc876c3e | ||
|
|
10fc78c869 | ||
|
|
9fa66ac80f | ||
|
|
974a4af7f4 | ||
|
|
9026c56ebf | ||
|
|
26ca9599de |
@@ -38,11 +38,16 @@ For smaller displays you may wish to enable `single_pane_mode`:
|
||||
|
||||
## Commands
|
||||
|
||||
- `CTRL` + `k` = display a list of commands.
|
||||
- `↑→↓←` = Navigate around the UI.
|
||||
- `F1/F2/F3` = Jump to Channel/Messages/Nodes
|
||||
- `ENTER` = Send a message typed in the Input Window, or with the Node List highlighted, select a node to DM
|
||||
- `` ` `` = Open the Settings dialogue
|
||||
- `` ` `` or `F12` = Open the Settings dialogue
|
||||
- `CTRL` + `p` = Hide/show a log of raw received packets.
|
||||
- `CTRL` + `t` = With the Node List highlighted, send a traceroute to the selected node
|
||||
- `CTRL` + `t` or `F4` = With the Node List highlighted, send a traceroute to the selected node
|
||||
- `F5` = Display a node's info
|
||||
- `CTRL` + `f` = With the Node List highlighted, favorite the selected node
|
||||
- `CTRL` + `g` = With the Node List highlighted, ignore the selected node
|
||||
- `CTRL` + `d` = With the Channel List hightlighted, archive a chat to reduce UI clutter. Messages will be saved in the db and repopulate if you send or receive a DM from this user.
|
||||
- `CTRL` + `d` = With the Note List highlghted, remove a node from your nodedb.
|
||||
- `ESC` = Exit out of the Settings Dialogue, or Quit the application if settings are not displayed.
|
||||
|
||||
@@ -129,8 +129,10 @@ def start() -> None:
|
||||
|
||||
try:
|
||||
curses.wrapper(main)
|
||||
interface_state.interface.close()
|
||||
except KeyboardInterrupt:
|
||||
logging.info("User exited with Ctrl+C")
|
||||
interface_state.interface.close()
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
logging.critical("Fatal error", exc_info=True)
|
||||
|
||||
@@ -2,7 +2,46 @@ import logging
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import time
|
||||
import subprocess
|
||||
import threading
|
||||
# Debounce notification sounds so a burst of queued messages only plays once.
|
||||
_SOUND_DEBOUNCE_SECONDS = 0.8
|
||||
_sound_timer: threading.Timer | None = None
|
||||
_sound_timer_lock = threading.Lock()
|
||||
_last_sound_request = 0.0
|
||||
|
||||
|
||||
def schedule_notification_sound(delay: float = _SOUND_DEBOUNCE_SECONDS) -> None:
|
||||
"""Schedule a notification sound after a short quiet period.
|
||||
|
||||
If more messages arrive before the delay elapses, the timer is reset.
|
||||
This prevents playing a sound for each message when a backlog flushes.
|
||||
"""
|
||||
global _sound_timer, _last_sound_request
|
||||
|
||||
now = time.monotonic()
|
||||
with _sound_timer_lock:
|
||||
_last_sound_request = now
|
||||
|
||||
# Cancel any previously scheduled sound.
|
||||
if _sound_timer is not None:
|
||||
try:
|
||||
_sound_timer.cancel()
|
||||
except Exception:
|
||||
pass
|
||||
_sound_timer = None
|
||||
|
||||
def _fire(expected_request_time: float) -> None:
|
||||
# Only play if nothing newer has been scheduled.
|
||||
with _sound_timer_lock:
|
||||
if expected_request_time != _last_sound_request:
|
||||
return
|
||||
play_sound()
|
||||
|
||||
_sound_timer = threading.Timer(delay, _fire, args=(now,))
|
||||
_sound_timer.daemon = True
|
||||
_sound_timer.start()
|
||||
from typing import Any, Dict
|
||||
|
||||
from contact.utilities.utils import (
|
||||
@@ -101,9 +140,14 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
maybe_store_nodeinfo_in_db(packet)
|
||||
|
||||
elif packet["decoded"]["portnum"] == "TEXT_MESSAGE_APP":
|
||||
hop_start = packet.get('hopStart', 0)
|
||||
hop_limit = packet.get('hopLimit', 0)
|
||||
|
||||
hops = hop_start - hop_limit
|
||||
|
||||
|
||||
if config.notification_sound == "True":
|
||||
play_sound()
|
||||
schedule_notification_sound()
|
||||
|
||||
message_bytes = packet["decoded"]["payload"]
|
||||
message_string = message_bytes.decode("utf-8")
|
||||
@@ -140,7 +184,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
message_from_id = packet["from"]
|
||||
message_from_string = get_name_from_database(message_from_id, type="short") + ":"
|
||||
|
||||
add_new_message(channel_id, f"{config.message_prefix} {message_from_string} ", message_string)
|
||||
add_new_message(channel_id, f"{config.message_prefix} [{hops}] {message_from_string} ", message_string)
|
||||
|
||||
if refresh_channels:
|
||||
draw_channel_list()
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import time
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
import google.protobuf.json_format
|
||||
@@ -49,7 +51,7 @@ def onAckNak(packet: Dict[str, Any]) -> None:
|
||||
ack_type = "Nak"
|
||||
|
||||
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
|
||||
config.sent_message_prefix + confirm_string + ": ",
|
||||
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
|
||||
message,
|
||||
)
|
||||
|
||||
|
||||
@@ -63,7 +63,7 @@ def paint_frame(win, selected: bool) -> None:
|
||||
|
||||
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
"""Handle terminal resize events and redraw the UI accordingly."""
|
||||
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, function_win, packetlog_win, entry_win
|
||||
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, packetlog_win, entry_win
|
||||
|
||||
# Calculate window max dimensions
|
||||
height, width = stdscr.getmaxyx()
|
||||
@@ -91,20 +91,18 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
nodes_width = max(MIN_COL, nodes_width - delta)
|
||||
|
||||
entry_height = 3
|
||||
function_height = 3
|
||||
y_pad = entry_height + function_height
|
||||
y_pad = entry_height
|
||||
content_h = max(1, height - y_pad)
|
||||
pkt_h = max(1, int(height / 3))
|
||||
|
||||
if firstrun:
|
||||
entry_win = curses.newwin(entry_height, width, 0, 0)
|
||||
entry_win = curses.newwin(entry_height, width, height - entry_height, 0)
|
||||
|
||||
channel_win = curses.newwin(content_h, channel_width, entry_height, 0)
|
||||
messages_win = curses.newwin(content_h, messages_width, entry_height, channel_width)
|
||||
nodes_win = curses.newwin(content_h, nodes_width, entry_height, channel_width + messages_width)
|
||||
channel_win = curses.newwin(content_h, channel_width, 0, 0)
|
||||
messages_win = curses.newwin(content_h, messages_width, 0, channel_width)
|
||||
nodes_win = curses.newwin(content_h, nodes_width, 0, channel_width + messages_width)
|
||||
|
||||
function_win = curses.newwin(function_height, width, height - function_height, 0)
|
||||
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - function_height, channel_width)
|
||||
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, channel_width)
|
||||
|
||||
# Will be resized to what we need when drawn
|
||||
messages_pad = curses.newpad(1, 1)
|
||||
@@ -112,7 +110,7 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
channel_pad = curses.newpad(1, 1)
|
||||
|
||||
# Set background colors for windows
|
||||
for win in [entry_win, channel_win, messages_win, nodes_win, function_win, packetlog_win]:
|
||||
for win in [entry_win, channel_win, messages_win, nodes_win, packetlog_win]:
|
||||
win.bkgd(get_color("background"))
|
||||
|
||||
# Set background colors for pads
|
||||
@@ -120,31 +118,30 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
pad.bkgd(get_color("background"))
|
||||
|
||||
# Set colors for window frames
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win]:
|
||||
win.attrset(get_color("window_frame"))
|
||||
|
||||
else:
|
||||
for win in [entry_win, channel_win, messages_win, nodes_win, function_win, packetlog_win]:
|
||||
for win in [entry_win, channel_win, messages_win, nodes_win, packetlog_win]:
|
||||
win.erase()
|
||||
|
||||
entry_win.resize(3, width)
|
||||
entry_win.resize(entry_height, width)
|
||||
entry_win.mvwin(height - entry_height, 0)
|
||||
|
||||
channel_win.resize(content_h, channel_width)
|
||||
channel_win.mvwin(0, 0)
|
||||
|
||||
messages_win.resize(content_h, messages_width)
|
||||
messages_win.mvwin(3, channel_width)
|
||||
messages_win.mvwin(0, channel_width)
|
||||
|
||||
nodes_win.resize(content_h, nodes_width)
|
||||
nodes_win.mvwin(entry_height, channel_width + messages_width)
|
||||
|
||||
function_win.resize(3, width)
|
||||
function_win.mvwin(height - function_height, 0)
|
||||
nodes_win.mvwin(0, channel_width + messages_width)
|
||||
|
||||
packetlog_win.resize(pkt_h, messages_width)
|
||||
packetlog_win.mvwin(height - pkt_h - function_height, channel_width)
|
||||
packetlog_win.mvwin(height - pkt_h - entry_height, channel_width)
|
||||
|
||||
# Draw window borders
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win]:
|
||||
win.box()
|
||||
win.refresh()
|
||||
|
||||
@@ -152,7 +149,6 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
curses.curs_set(1)
|
||||
|
||||
try:
|
||||
draw_function_win()
|
||||
draw_channel_list()
|
||||
draw_messages_window(True)
|
||||
draw_node_list()
|
||||
@@ -176,7 +172,7 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
handle_resize(stdscr, True)
|
||||
|
||||
while True:
|
||||
draw_text_field(entry_win, f"Input: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
|
||||
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
|
||||
|
||||
# Get user input from entry window
|
||||
char = entry_win.get_wch()
|
||||
@@ -204,16 +200,22 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
elif char == curses.KEY_LEFT or char == curses.KEY_RIGHT:
|
||||
handle_leftright(char)
|
||||
|
||||
elif char in (curses.KEY_F1, curses.KEY_F2, curses.KEY_F3):
|
||||
handle_function_keys(char)
|
||||
|
||||
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
|
||||
input_text = handle_enter(input_text)
|
||||
|
||||
elif char == chr(20): # Ctrl + t for Traceroute
|
||||
elif char in (curses.KEY_F4, chr(20)): # Ctrl + t and F4 for Traceroute
|
||||
handle_ctrl_t(stdscr)
|
||||
|
||||
elif char == curses.KEY_F5:
|
||||
handle_f5_key(stdscr)
|
||||
|
||||
elif char in (curses.KEY_BACKSPACE, chr(127)):
|
||||
input_text = handle_backspace(entry_win, input_text)
|
||||
|
||||
elif char == "`": # ` Launch the settings interface
|
||||
elif char in (curses.KEY_F12, "`"): # ` Launch the settings interface
|
||||
handle_backtick(stdscr)
|
||||
|
||||
elif char == chr(16): # Ctrl + P for Packet Log
|
||||
@@ -229,6 +231,9 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
elif char == chr(31): # Ctrl + / to search
|
||||
handle_ctrl_fslash()
|
||||
|
||||
elif char == chr(11): # Ctrl + K for Help
|
||||
handle_ctrl_k(stdscr)
|
||||
|
||||
elif char == chr(6): # Ctrl + F to toggle favorite
|
||||
handle_ctrl_f(stdscr)
|
||||
|
||||
@@ -336,7 +341,6 @@ def handle_leftright(char: int) -> None:
|
||||
paint_frame(messages_win, selected=False)
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
draw_function_win()
|
||||
paint_frame(nodes_win, selected=False)
|
||||
refresh_pad(2)
|
||||
|
||||
@@ -350,11 +354,54 @@ def handle_leftright(char: int) -> None:
|
||||
paint_frame(messages_win, selected=True)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
draw_function_win()
|
||||
paint_frame(nodes_win, selected=True)
|
||||
refresh_pad(2)
|
||||
|
||||
# Draw arrows last; force even in multi-pane to avoid flicker
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
|
||||
def handle_function_keys(char: int) -> None:
|
||||
"""Switch windows using F1/F2/F3."""
|
||||
if char == curses.KEY_F1:
|
||||
target = 0
|
||||
elif char == curses.KEY_F2:
|
||||
target = 1
|
||||
elif char == curses.KEY_F3:
|
||||
target = 2
|
||||
else:
|
||||
return
|
||||
|
||||
old_window = ui_state.current_window
|
||||
|
||||
if target == old_window:
|
||||
return
|
||||
|
||||
ui_state.current_window = target
|
||||
handle_resize(root_win, False)
|
||||
|
||||
if old_window == 0:
|
||||
paint_frame(channel_win, selected=False)
|
||||
refresh_pad(0)
|
||||
elif old_window == 1:
|
||||
paint_frame(messages_win, selected=False)
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
paint_frame(nodes_win, selected=False)
|
||||
refresh_pad(2)
|
||||
|
||||
if not ui_state.single_pane_mode:
|
||||
draw_window_arrows(old_window)
|
||||
|
||||
if ui_state.current_window == 0:
|
||||
paint_frame(channel_win, selected=True)
|
||||
refresh_pad(0)
|
||||
elif ui_state.current_window == 1:
|
||||
paint_frame(messages_win, selected=True)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
paint_frame(nodes_win, selected=True)
|
||||
refresh_pad(2)
|
||||
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
|
||||
@@ -402,13 +449,111 @@ def handle_enter(input_text: str) -> str:
|
||||
return input_text
|
||||
|
||||
|
||||
def handle_f5_key(stdscr: curses.window) -> None:
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
message_parts = []
|
||||
|
||||
message_parts.append("**📋 Basic Information:**")
|
||||
message_parts.append(f"• Device: {node.get('user', {}).get('longName', 'Unknown')}")
|
||||
message_parts.append(f"• Short name: {node.get('user', {}).get('shortName', 'Unknown')}")
|
||||
message_parts.append(f"• Hardware: {node.get('user', {}).get('hwModel', 'Unknown')}")
|
||||
|
||||
role = f"{node.get('user', {}).get('role', 'Unknown')}"
|
||||
message_parts.append(f"• Role: {role}")
|
||||
|
||||
pk = f"{node.get('user', {}).get('publicKey')}"
|
||||
message_parts.append(f"Public key: {pk}")
|
||||
|
||||
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
|
||||
if "position" in node:
|
||||
pos = node["position"]
|
||||
if pos.get("latitude") and pos.get("longitude"):
|
||||
message_parts.append(f"• Position: {pos['latitude']:.4f}, {pos['longitude']:.4f}")
|
||||
if pos.get("altitude"):
|
||||
message_parts.append(f"• Altitude: {pos['altitude']}m")
|
||||
message_parts.append(f"https://maps.google.com/?q={pos['latitude']:.4f},{pos['longitude']:.4f}")
|
||||
|
||||
if any(key in node for key in ["snr", "hopsAway", "lastHeard"]):
|
||||
message_parts.append("\n**🌐 Network Metrics:**")
|
||||
|
||||
if "snr" in node:
|
||||
snr = node["snr"]
|
||||
snr_status = (
|
||||
"🟢 Excellent"
|
||||
if snr > 10
|
||||
else (
|
||||
"🟡 Good"
|
||||
if snr > 3
|
||||
else "🟠 Fair" if snr > -10 else "🔴 Poor" if snr > -20 else "💀 Very Poor"
|
||||
)
|
||||
)
|
||||
message_parts.append(f"• SNR: {snr}dB {snr_status}")
|
||||
|
||||
if "hopsAway" in node:
|
||||
hops = node["hopsAway"]
|
||||
hop_emoji = "📡" if hops == 0 else "🔄" if hops == 1 else "⏩"
|
||||
message_parts.append(f"• Hops away: {hop_emoji} {hops}")
|
||||
|
||||
if "lastHeard" in node and node["lastHeard"]:
|
||||
message_parts.append(f"• Last heard: 🕐 {get_time_ago(node['lastHeard'])}")
|
||||
|
||||
if node.get("deviceMetrics"):
|
||||
metrics = node["deviceMetrics"]
|
||||
message_parts.append("\n**📊 Device Metrics:**")
|
||||
|
||||
if "batteryLevel" in metrics:
|
||||
battery = metrics["batteryLevel"]
|
||||
battery_emoji = "🟢" if battery > 50 else "🟡" if battery > 20 else "🔴"
|
||||
voltage_info = f" ({metrics['voltage']}v)" if "voltage" in metrics else ""
|
||||
message_parts.append(f"• Battery: {battery_emoji} {battery}%{voltage_info}")
|
||||
|
||||
if "uptimeSeconds" in metrics:
|
||||
message_parts.append(f"• Uptime: ⏱️ {get_readable_duration(metrics['uptimeSeconds'])}")
|
||||
|
||||
if "channelUtilization" in metrics:
|
||||
util = metrics["channelUtilization"]
|
||||
util_emoji = "🔴" if util > 80 else "🟡" if util > 50 else "🟢"
|
||||
message_parts.append(f"• Channel utilization: {util_emoji} {util:.2f}%")
|
||||
|
||||
if "airUtilTx" in metrics:
|
||||
air_util = metrics["airUtilTx"]
|
||||
air_emoji = "🔴" if air_util > 80 else "🟡" if air_util > 50 else "🟢"
|
||||
message_parts.append(f"• Air utilization TX: {air_emoji} {air_util:.2f}%")
|
||||
|
||||
message = "\n".join(message_parts)
|
||||
|
||||
contact.ui.dialog.dialog(f"📡 Node Details: {node.get('user', {}).get('shortName', 'Unknown')}", message)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
|
||||
def handle_ctrl_t(stdscr: curses.window) -> None:
|
||||
"""Handle Ctrl + T key events to send a traceroute."""
|
||||
now = time.monotonic()
|
||||
cooldown = 30.0
|
||||
remaining = cooldown - (now - ui_state.last_traceroute_time)
|
||||
|
||||
if remaining > 0:
|
||||
curses.curs_set(0) # Hide cursor
|
||||
contact.ui.dialog.dialog(
|
||||
"Traceroute Not Sent", f"Please wait {int(remaining)} seconds before sending another traceroute."
|
||||
)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
return
|
||||
|
||||
send_traceroute()
|
||||
ui_state.last_traceroute_time = now
|
||||
curses.curs_set(0) # Hide cursor
|
||||
contact.ui.dialog.dialog(
|
||||
f"Traceroute Sent To: {get_name_from_database(ui_state.node_list[ui_state.selected_node])}",
|
||||
"Results will appear in messages window.\nNote: Traceroute is limited to once every 30 seconds.",
|
||||
"Results will appear in messages window.",
|
||||
)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
@@ -450,6 +595,34 @@ def handle_ctrl_p() -> None:
|
||||
draw_messages_window(True)
|
||||
|
||||
|
||||
# --- Ctrl+K handler for Help ---
|
||||
def handle_ctrl_k(stdscr: curses.window) -> None:
|
||||
"""Handle Ctrl + K to show a help window with shortcut keys."""
|
||||
curses.curs_set(0)
|
||||
|
||||
cmds = [
|
||||
"↑/↓ = Scroll",
|
||||
"←/→ = Switch window",
|
||||
"F1/F2/F3 = Jump to Channel/Messages/Nodes",
|
||||
"ENTER = Send / Select",
|
||||
"` or F12 = Settings",
|
||||
"ESC = Quit",
|
||||
"Ctrl+P = Toggle Packet Log",
|
||||
"Ctrl+T or F4 = Traceroute",
|
||||
"F5 = Full node info",
|
||||
"Ctrl+D = Archive chat / remove node",
|
||||
"Ctrl+F = Favorite",
|
||||
"Ctrl+G = Ignore",
|
||||
"Ctrl+/ = Search",
|
||||
"Ctrl+K = Help",
|
||||
]
|
||||
|
||||
contact.ui.dialog.dialog("Help — Shortcut Keys", "\n".join(cmds))
|
||||
|
||||
curses.curs_set(1)
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def handle_ctrl_d() -> None:
|
||||
if ui_state.current_window == 0:
|
||||
if isinstance(ui_state.channel_list[ui_state.selected_channel], int):
|
||||
@@ -634,7 +807,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
|
||||
for line in wrapped_lines:
|
||||
if prefix.startswith("--"):
|
||||
color = get_color("timestamps")
|
||||
elif prefix.startswith(config.sent_message_prefix):
|
||||
elif prefix.find(config.sent_message_prefix) != -1:
|
||||
color = get_color("tx_messages")
|
||||
else:
|
||||
color = get_color("rx_messages")
|
||||
@@ -684,9 +857,21 @@ def draw_node_list() -> None:
|
||||
for i, node_num in enumerate(ui_state.node_list):
|
||||
node = interface_state.interface.nodesByNum[node_num]
|
||||
secure = "user" in node and "publicKey" in node["user"] and node["user"]["publicKey"]
|
||||
node_str = f"{'🔐' if secure else '🔓'} {get_name_from_database(node_num, 'long')}".ljust(box_width - 2)[
|
||||
: box_width - 2
|
||||
]
|
||||
status_icon = "🔐" if secure else "🔓"
|
||||
node_name = get_name_from_database(node_num, "long")
|
||||
user_name = node["user"]["shortName"]
|
||||
|
||||
uptime_str = ""
|
||||
if "deviceMetrics" in node and "uptimeSeconds" in node["deviceMetrics"]:
|
||||
uptime_str = f" / Up: {get_readable_duration(node['deviceMetrics']['uptimeSeconds'])}"
|
||||
|
||||
last_heard_str = f" ■ {get_time_ago(node['lastHeard'])}" if node.get("lastHeard") else ""
|
||||
hops_str = f" ■ Hops: {node['hopsAway']}" if "hopsAway" in node else ""
|
||||
snr_str = f" ■ SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
|
||||
|
||||
# Future node name custom formatting possible
|
||||
node_str = f"{status_icon} {node_name}"
|
||||
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
|
||||
color = "node_list"
|
||||
if "isFavorite" in node and node["isFavorite"]:
|
||||
color = "node_favorite"
|
||||
@@ -786,8 +971,6 @@ def select_node(idx: int) -> None:
|
||||
ui_state=ui_state,
|
||||
)
|
||||
|
||||
draw_function_win()
|
||||
|
||||
|
||||
def scroll_nodes(direction: int) -> None:
|
||||
"""Scroll through the node list by a given direction."""
|
||||
@@ -904,100 +1087,12 @@ def search(win: int) -> None:
|
||||
entry_win.erase()
|
||||
|
||||
|
||||
def draw_node_details() -> None:
|
||||
"""Draw the details of the selected node in the function window."""
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
except KeyError:
|
||||
def refresh_pad(window: int) -> None:
|
||||
|
||||
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
|
||||
if ui_state.single_pane_mode and window != ui_state.current_window:
|
||||
return
|
||||
|
||||
function_win.erase()
|
||||
function_win.box()
|
||||
|
||||
nodestr = ""
|
||||
width = function_win.getmaxyx()[1]
|
||||
|
||||
node_details_list = [
|
||||
f"{node['user']['longName']} " if "user" in node and "longName" in node["user"] else "",
|
||||
f"({node['user']['shortName']})" if "user" in node and "shortName" in node["user"] else "",
|
||||
f" | {node['user']['hwModel']}" if "user" in node and "hwModel" in node["user"] else "",
|
||||
f" | {node['user']['role']}" if "user" in node and "role" in node["user"] else "",
|
||||
]
|
||||
|
||||
if ui_state.node_list[ui_state.selected_node] == interface_state.myNodeNum:
|
||||
node_details_list.extend(
|
||||
[
|
||||
(
|
||||
f" | Bat: {node['deviceMetrics']['batteryLevel']}% ({node['deviceMetrics']['voltage']}v)"
|
||||
if "deviceMetrics" in node
|
||||
and "batteryLevel" in node["deviceMetrics"]
|
||||
and "voltage" in node["deviceMetrics"]
|
||||
else ""
|
||||
),
|
||||
(
|
||||
f" | Up: {get_readable_duration(node['deviceMetrics']['uptimeSeconds'])}"
|
||||
if "deviceMetrics" in node and "uptimeSeconds" in node["deviceMetrics"]
|
||||
else ""
|
||||
),
|
||||
(
|
||||
f" | ChUtil: {node['deviceMetrics']['channelUtilization']:.2f}%"
|
||||
if "deviceMetrics" in node and "channelUtilization" in node["deviceMetrics"]
|
||||
else ""
|
||||
),
|
||||
(
|
||||
f" | AirUtilTX: {node['deviceMetrics']['airUtilTx']:.2f}%"
|
||||
if "deviceMetrics" in node and "airUtilTx" in node["deviceMetrics"]
|
||||
else ""
|
||||
),
|
||||
]
|
||||
)
|
||||
else:
|
||||
node_details_list.extend(
|
||||
[
|
||||
f" | {get_time_ago(node['lastHeard'])}" if ("lastHeard" in node and node["lastHeard"]) else "",
|
||||
f" | Hops: {node['hopsAway']}" if "hopsAway" in node else "",
|
||||
f" | SNR: {node['snr']}dB" if ("snr" in node and "hopsAway" in node and node["hopsAway"] == 0) else "",
|
||||
]
|
||||
)
|
||||
|
||||
for s in node_details_list:
|
||||
if len(nodestr) + len(s) < width - 2:
|
||||
nodestr = nodestr + s
|
||||
|
||||
draw_centered_text_field(function_win, nodestr, 0, get_color("commands"))
|
||||
|
||||
|
||||
def draw_help() -> None:
|
||||
"""Draw the help text in the function window."""
|
||||
cmds = [
|
||||
"↑→↓← = Select",
|
||||
" ENTER = Send",
|
||||
" ` = Settings",
|
||||
" ESC = Quit",
|
||||
" ^P = Packet Log",
|
||||
" ^t = Traceroute",
|
||||
" ^d = Archive Chat",
|
||||
" ^f = Favorite",
|
||||
" ^g = Ignore",
|
||||
" ^/ = Search",
|
||||
]
|
||||
function_str = ""
|
||||
for s in cmds:
|
||||
if len(function_str) + len(s) < function_win.getmaxyx()[1] - 2:
|
||||
function_str += s
|
||||
|
||||
draw_centered_text_field(function_win, function_str, 0, get_color("commands"))
|
||||
|
||||
|
||||
def draw_function_win() -> None:
|
||||
if ui_state.current_window == 2:
|
||||
draw_node_details()
|
||||
else:
|
||||
draw_help()
|
||||
|
||||
|
||||
def refresh_pad(window: int) -> None:
|
||||
# Derive the target box and pad for the requested window
|
||||
win_height = channel_win.getmaxyx()[0]
|
||||
|
||||
@@ -1016,6 +1111,7 @@ def refresh_pad(window: int) -> None:
|
||||
pad = nodes_pad
|
||||
box = nodes_win
|
||||
lines = box.getmaxyx()[0] - 2
|
||||
box.addstr(0, 2, (f"Nodes: {len(ui_state.node_list)}"), curses.A_BOLD)
|
||||
selected_item = ui_state.selected_node
|
||||
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
|
||||
|
||||
@@ -1026,10 +1122,6 @@ def refresh_pad(window: int) -> None:
|
||||
selected_item = ui_state.selected_channel
|
||||
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
|
||||
|
||||
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
|
||||
if ui_state.single_pane_mode and window != ui_state.current_window:
|
||||
return
|
||||
|
||||
# Compute inner drawable area of the box
|
||||
box_y, box_x = box.getbegyx()
|
||||
box_h, box_w = box.getmaxyx()
|
||||
@@ -1078,7 +1170,23 @@ def remove_notification(channel_number: int) -> None:
|
||||
|
||||
def draw_text_field(win: curses.window, text: str, color: int) -> None:
|
||||
win.border()
|
||||
win.addstr(1, 1, text, color)
|
||||
|
||||
# Put a small hint in the border of the message entry field.
|
||||
# We key off the "Message:" prompt to avoid affecting other bordered fields.
|
||||
if isinstance(text, str) and text.startswith("Message:"):
|
||||
hint = " Ctrl+K Help "
|
||||
h, w = win.getmaxyx()
|
||||
x = max(2, w - len(hint) - 2)
|
||||
try:
|
||||
win.addstr(0, x, hint, get_color("commands"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Draw the actual field text
|
||||
try:
|
||||
win.addstr(1, 1, text, color)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
|
||||
def draw_centered_text_field(win: curses.window, text: str, y_offset: int, color: int) -> None:
|
||||
@@ -1087,8 +1195,3 @@ def draw_centered_text_field(win: curses.window, text: str, y_offset: int, color
|
||||
y = (height // 2) + y_offset
|
||||
win.addstr(y, x, text, color)
|
||||
win.refresh()
|
||||
|
||||
|
||||
def draw_debug(value: Union[str, int]) -> None:
|
||||
function_win.addstr(1, 1, f"debug: {value} ")
|
||||
function_win.refresh()
|
||||
|
||||
@@ -55,10 +55,12 @@ field_mapping, help_text = parse_ini_file(translation_file)
|
||||
|
||||
|
||||
def display_menu() -> tuple[object, object]:
|
||||
if help_win:
|
||||
min_help_window_height = 6
|
||||
else:
|
||||
min_help_window_height = 0
|
||||
# if help_win:
|
||||
# min_help_window_height = 6
|
||||
# else:
|
||||
# min_help_window_height = 0
|
||||
|
||||
min_help_window_height = 6
|
||||
|
||||
num_items = len(menu_state.current_menu) + (1 if menu_state.show_save_option else 0)
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import curses
|
||||
from contact.ui.colors import get_color
|
||||
from contact.ui.nav_utils import draw_main_arrows
|
||||
from contact.utilities.singleton import menu_state, ui_state
|
||||
|
||||
|
||||
@@ -13,12 +14,40 @@ def dialog(title: str, message: str) -> None:
|
||||
height, width = curses.LINES, curses.COLS
|
||||
|
||||
# Parse message into lines and calculate dimensions
|
||||
message_lines = message.splitlines()
|
||||
message_lines = message.splitlines() or [""]
|
||||
max_line_length = max(len(l) for l in message_lines)
|
||||
|
||||
# Desired size
|
||||
dialog_width = max(len(title) + 4, max_line_length + 4)
|
||||
dialog_height = len(message_lines) + 4
|
||||
x = (width - dialog_width) // 2
|
||||
y = (height - dialog_height) // 2
|
||||
desired_height = len(message_lines) + 4
|
||||
|
||||
# Clamp dialog size to the screen (leave a 1-cell margin if possible)
|
||||
max_w = max(10, width - 2)
|
||||
max_h = max(6, height - 2)
|
||||
dialog_width = min(dialog_width, max_w)
|
||||
dialog_height = min(desired_height, max_h)
|
||||
|
||||
x = max(0, (width - dialog_width) // 2)
|
||||
y = max(0, (height - dialog_height) // 2)
|
||||
|
||||
# Ensure we have a start index slot for this dialog window id (4)
|
||||
# ui_state.start_index is used by draw_main_arrows()
|
||||
try:
|
||||
while len(ui_state.start_index) <= 4:
|
||||
ui_state.start_index.append(0)
|
||||
except Exception:
|
||||
# If start_index isn't list-like, fall back to an attribute
|
||||
if not hasattr(ui_state, "start_index"):
|
||||
ui_state.start_index = [0, 0, 0, 0, 0]
|
||||
|
||||
def visible_message_rows() -> int:
|
||||
# Rows available for message text inside the border, excluding title row and OK row.
|
||||
# Layout:
|
||||
# row 0: title
|
||||
# rows 1..(dialog_height-3): message viewport (with arrows drawn on a subwindow)
|
||||
# row dialog_height-2: OK button
|
||||
# So message viewport height is dialog_height - 3 - 1 + 1 = dialog_height - 3
|
||||
return max(1, dialog_height - 4)
|
||||
|
||||
def draw_window():
|
||||
win.erase()
|
||||
@@ -26,23 +55,66 @@ def dialog(title: str, message: str) -> None:
|
||||
win.attrset(get_color("window_frame"))
|
||||
win.border(0)
|
||||
|
||||
win.addstr(0, 2, title, get_color("settings_default"))
|
||||
# Title
|
||||
try:
|
||||
win.addstr(0, 2, title[: max(0, dialog_width - 4)], get_color("settings_default"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
for i, line in enumerate(message_lines):
|
||||
msg_x = (dialog_width - len(line)) // 2
|
||||
win.addstr(2 + i, msg_x, line, get_color("settings_default"))
|
||||
# Message viewport
|
||||
viewport_h = visible_message_rows()
|
||||
start = ui_state.start_index[4]
|
||||
start = max(0, min(start, max(0, len(message_lines) - viewport_h)))
|
||||
ui_state.start_index[4] = start
|
||||
|
||||
# Create a subwindow covering the message region so draw_main_arrows() doesn't collide with the OK row
|
||||
msg_win = win.derwin(viewport_h + 2, dialog_width - 2, 1, 1)
|
||||
msg_win.erase()
|
||||
|
||||
for i in range(viewport_h):
|
||||
idx = start + i
|
||||
if idx >= len(message_lines):
|
||||
break
|
||||
line = message_lines[idx]
|
||||
# Hard-trim lines that don't fit
|
||||
trimmed = line[: max(0, dialog_width - 6)]
|
||||
msg_x = max(0, ((dialog_width - 2) - len(trimmed)) // 2)
|
||||
try:
|
||||
msg_win.addstr(1 + i, msg_x, trimmed, get_color("settings_default"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Draw arrows only when scrolling is needed
|
||||
if len(message_lines) > viewport_h:
|
||||
draw_main_arrows(msg_win, len(message_lines) - 1, window=4)
|
||||
else:
|
||||
# Clear arrow positions if not needed
|
||||
try:
|
||||
h, w = msg_win.getmaxyx()
|
||||
msg_win.addstr(1, w - 2, " ", get_color("settings_default"))
|
||||
msg_win.addstr(h - 2, w - 2, " ", get_color("settings_default"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
msg_win.noutrefresh()
|
||||
|
||||
# OK button
|
||||
ok_text = " Ok "
|
||||
win.addstr(
|
||||
dialog_height - 2,
|
||||
(dialog_width - len(ok_text)) // 2,
|
||||
ok_text,
|
||||
get_color("settings_default", reverse=True),
|
||||
)
|
||||
try:
|
||||
win.addstr(
|
||||
dialog_height - 2,
|
||||
(dialog_width - len(ok_text)) // 2,
|
||||
ok_text,
|
||||
get_color("settings_default", reverse=True),
|
||||
)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
win.refresh()
|
||||
win.noutrefresh()
|
||||
curses.doupdate()
|
||||
|
||||
win = curses.newwin(dialog_height, dialog_width, y, x)
|
||||
win.keypad(True)
|
||||
draw_window()
|
||||
|
||||
while True:
|
||||
@@ -51,9 +123,19 @@ def dialog(title: str, message: str) -> None:
|
||||
|
||||
if menu_state.need_redraw:
|
||||
menu_state.need_redraw = False
|
||||
curses.update_lines_cols()
|
||||
height, width = curses.LINES, curses.COLS
|
||||
draw_window()
|
||||
|
||||
if char in (curses.KEY_ENTER, 10, 13, 32, 27): # Enter, space, Esc
|
||||
# Close dialog
|
||||
ok_selected = True
|
||||
if char in (27, curses.KEY_LEFT): # Esc or Left arrow
|
||||
win.erase()
|
||||
win.refresh()
|
||||
ui_state.current_window = previous_window
|
||||
return
|
||||
|
||||
if ok_selected and char in (curses.KEY_ENTER, 10, 13, 32):
|
||||
win.erase()
|
||||
win.refresh()
|
||||
ui_state.current_window = previous_window
|
||||
@@ -61,3 +143,22 @@ def dialog(title: str, message: str) -> None:
|
||||
|
||||
if char == -1:
|
||||
continue
|
||||
|
||||
# Scroll if the dialog is clipped vertically
|
||||
viewport_h = visible_message_rows()
|
||||
if len(message_lines) > viewport_h:
|
||||
start = ui_state.start_index[4]
|
||||
max_start = max(0, len(message_lines) - viewport_h)
|
||||
|
||||
if char in (curses.KEY_UP, ord("k")):
|
||||
ui_state.start_index[4] = max(0, start - 1)
|
||||
draw_window()
|
||||
elif char in (curses.KEY_DOWN, ord("j")):
|
||||
ui_state.start_index[4] = min(max_start, start + 1)
|
||||
draw_window()
|
||||
elif char == curses.KEY_PPAGE: # Page up
|
||||
ui_state.start_index[4] = max(0, start - viewport_h)
|
||||
draw_window()
|
||||
elif char == curses.KEY_NPAGE: # Page down
|
||||
ui_state.start_index[4] = min(max_start, start + viewport_h)
|
||||
draw_window()
|
||||
|
||||
@@ -26,6 +26,7 @@ class ChatUIState:
|
||||
selected_node: int = 0
|
||||
current_window: int = 0
|
||||
last_sent_time: float = 0.0
|
||||
last_traceroute_time: float = 0.0
|
||||
|
||||
selected_index: int = 0
|
||||
start_index: List[int] = field(default_factory=lambda: [0, 0, 0])
|
||||
|
||||
@@ -136,13 +136,18 @@ def load_messages_from_db() -> None:
|
||||
elif ack_type == "Nak":
|
||||
ack_str = config.nak_str
|
||||
|
||||
ts_str = datetime.fromtimestamp(timestamp).strftime("[%H:%M:%S]")
|
||||
|
||||
if user_id == str(interface_state.myNodeNum):
|
||||
sanitized_message = message.replace("\x00", "")
|
||||
formatted_message = (f"{config.sent_message_prefix}{ack_str}: ", sanitized_message)
|
||||
formatted_message = (
|
||||
f"{ts_str} {config.sent_message_prefix}{ack_str}: ",
|
||||
sanitized_message,
|
||||
)
|
||||
else:
|
||||
sanitized_message = message.replace("\x00", "")
|
||||
formatted_message = (
|
||||
f"{config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
|
||||
f"{ts_str} {config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
|
||||
sanitized_message,
|
||||
)
|
||||
|
||||
|
||||
@@ -167,7 +167,8 @@ def add_new_message(channel_id, prefix, message):
|
||||
ui_state.all_messages[channel_id].append((f"-- {current_hour} --", ""))
|
||||
|
||||
# Add the message
|
||||
ui_state.all_messages[channel_id].append((prefix, message))
|
||||
ts_str = time.strftime("[%H:%M:%S] ")
|
||||
ui_state.all_messages[channel_id].append((f"{ts_str}{prefix}", message))
|
||||
|
||||
|
||||
def parse_protobuf(packet: dict) -> Union[str, dict]:
|
||||
@@ -181,15 +182,12 @@ def parse_protobuf(packet: dict) -> Union[str, dict]:
|
||||
return payload
|
||||
|
||||
# These portnumbers carry information visible elswhere in the app, so we just note them in the logs
|
||||
match portnum:
|
||||
case "TEXT_MESSAGE_APP":
|
||||
return "✉️"
|
||||
case "NODEINFO_APP":
|
||||
return "Name identification payload"
|
||||
case "TRACEROUTE_APP":
|
||||
return "Traceroute payload"
|
||||
case _:
|
||||
pass
|
||||
if portnum == "TEXT_MESSAGE_APP":
|
||||
return "✉️"
|
||||
elif portnum == "NODEINFO_APP":
|
||||
return "Name identification payload"
|
||||
elif portnum == "TRACEROUTE_APP":
|
||||
return "Traceroute payload"
|
||||
|
||||
handler = protocols.get(portnums_pb2.PortNum.Value(portnum)) if portnum is not None else None
|
||||
if handler is not None and handler.protobufFactory is not None:
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.4.2"
|
||||
version = "1.4.6"
|
||||
description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
|
||||
authors = [
|
||||
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
|
||||
|
||||
Reference in New Issue
Block a user