Compare commits

..

10 Commits

Author SHA1 Message Date
pdxlocations
cf4137347f refactor 2025-07-26 21:18:30 -07:00
pdxlocations
12f81f6af6 tweaks 2025-07-26 20:12:50 -07:00
pdxlocations
684d298fac check for selected_config 2025-07-26 20:01:59 -07:00
pdxlocations
4179a3f8d0 redraw input 2025-07-26 16:39:24 -07:00
pdxlocations
7a61808f47 fix positions 2025-07-26 00:55:41 -07:00
pdxlocations
a8680ac0ed changes 2025-07-26 00:31:49 -07:00
pdxlocations
818575939a automatic types 2025-07-25 19:02:22 -07:00
pdxlocations
acaad849b0 add rules 2025-07-25 18:40:37 -07:00
pdxlocations
0b25dda4af validation framework 2025-07-25 18:32:37 -07:00
pdxlocations
18329f0128 init 2025-07-25 00:18:51 -07:00
21 changed files with 501 additions and 1307 deletions

3
.gitignore vendored
View File

@@ -8,5 +8,4 @@ client.log
settings.log
config.json
default_config.log
dist/
.vscode/launch.json
dist/

11
.vscode/launch.json vendored
View File

@@ -1,22 +1,13 @@
{
"version": "0.1.0",
"configurations": [
{
"name": "Python Debugger: main",
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"cwd": "${workspaceFolder}",
"module": "contact.__main__",
"args": []
},
{
"name": "Python Debugger: tcp",
"type": "debugpy",
"request": "launch",
"cwd": "${workspaceFolder}",
"module": "contact.__main__",
"args": ["--host","192.168.86.69"]
}
]
}

View File

@@ -6,13 +6,6 @@
```bash
pip install contact
```
> [!NOTE]
> Windows users must also install:
>
> ```powershell
> pip install windows-curses
> ```
> because the built-in curses module is not available on Windows.
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.
@@ -32,24 +25,14 @@ All messages will saved in a SQLite DB and restored upon relaunch of the app. Y
By navigating to Settings -> App Settings, you may customize your UI's icons, colors, and more!
For smaller displays you may wish to enable `single_pane_mode`:
<img width="486" height="194" alt="Screenshot 2025-08-22 at 11 15 54PM" src="https://github.com/user-attachments/assets/447c5d30-0850-4a4f-b0d4-976e4c5e329d" />
## 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
- `` ` `` or `F12` = Open the Settings dialogue
- `` ` `` = Open the Settings dialogue
- `CTRL` + `p` = Hide/show a log of raw received packets.
- `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` + `t` = With the Node List highlighted, send a traceroute to 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.
### Search
@@ -79,17 +62,8 @@ If no connection arguments are specified, the client will attempt a serial conne
contact --port /dev/ttyUSB0
contact --host 192.168.1.1
contact --ble BlAddressOfDevice
contact --port COM3
```
To quickly connect to localhost, use:
```sh
contact -t
```
## Install in development (editable) mode:
```bash
git clone https://github.com/pdxlocations/contact.git
cd contact
python3 -m venv .venv
source .venv/bin/activate
pip install -e .
```

View File

@@ -72,7 +72,6 @@ def initialize_globals() -> None:
interface_state.myNodeNum = get_nodeNum()
ui_state.channel_list = get_channels()
ui_state.node_list = get_node_list()
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
pub.subscribe(on_receive, "meshtastic.receive")
init_nodedb()
@@ -103,9 +102,6 @@ def main(stdscr: curses.window) -> None:
initialize_globals()
logging.info("Starting main UI")
stdscr.clear()
stdscr.refresh()
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
main_ui(stdscr)
@@ -129,10 +125,8 @@ 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)

View File

@@ -14,13 +14,13 @@ id, "", ""
uplink_enabled, "Uplink enabled", "Let this channel's data be sent to the MQTT server configured on this node."
downlink_enabled, "Downlink enabled", "Let data from the MQTT server configured on this node be sent to this channel."
module_settings, "Module settings", "Position precision and Client Mute."
module_settings.position_precision, "Position precision", "The precision level of location data sent on this channel."
module_settings.is_client_muted, "Is Client Muted", "Controls whether or not the phone / clients should mute the current channel. Useful for noisy public channels you don't necessarily want to disable."
position_precision, "Position precision", "The precision level of location data sent on this channel."
is_client_muted, "", ""
[config.device]
title, "Device"
role, "Role", "For the vast majority of users, the correct choice is CLIENT. See Meshtastic docs for more information."
serial_enabled, "Enable serial console", "Serial Console over the Stream API."
serial_enabled, "Enable serial console", ""
button_gpio, "Button GPIO", "GPIO pin for user button."
buzzer_gpio, "Buzzer GPIO", "GPIO pin for user buzzer."
rebroadcast_mode, "Rebroadcast mode", "This setting defines the device's behavior for how messages are rebroadcast."
@@ -30,7 +30,6 @@ is_managed, "Enable managed mode", "Enabling Managed Mode blocks smartphone apps
disable_triple_click, "Disable triple button press", ""
tzdef, "Timezone", "Uses the TZ Database format to display the correct local time on the device display and in its logs."
led_heartbeat_disabled, "Disable LED heartbeat", "On certain hardware models, this disables the blinking heartbeat LED."
buzzer_mode, "Buzzer Mode", "Controls buzzer behavior for audio feedback."
[config.position]
title, "Position"
@@ -78,7 +77,6 @@ subnet, "IPv4 subnet", ""
dns, "IPv4 DNS server", ""
rsyslog_server, "RSyslog server", ""
enabled_protocols, "Enabled protocols", ""
ipv6_enabled, "IPv6 enabled", "Enables or Disables IPv6 networking."
[config.network.ipv4_config]
title, "IPv4 Config", ""
@@ -160,7 +158,7 @@ channel_num, "Frequency slot", "Determines the exact frequency the radio transmi
override_duty_cycle, "Override duty cycle", "Override the legal transmit time limit to allow unlimited transmit time. [warning]May have legal ramifications.[/warning]"
sx126x_rx_boosted_gain, "Enable SX126X RX boosted gain", "This is an option specific to the SX126x chip series which allows the chip to consume a small amount of additional power to increase RX (receive) sensitivity."
override_frequency, "Override frequency in MHz", "Overrides frequency slot. May have legal ramifications."
pa_fan_disabled, "PA Fan Disabled", "If true, disable the build-in PA FAN using pin define in RF95_FAN_EN"
pa_fan_disabled, "", ""
ignore_mqtt, "Ignore MQTT", "Ignores any messages it receives via LoRa that came via MQTT somewhere along the path towards the device."
config_ok_to_mqtt, "OK to MQTT", "Indicates that the user approves their packets to be uplinked to MQTT brokers."
@@ -192,10 +190,8 @@ tls_enabled, "TLS enabled", "If true, we attempt to establish a secure connectio
root, "Root topic", "The root topic to use for MQTT messages. This is useful if you want to use a single MQTT server for multiple meshtastic networks and separate them via ACLs."
proxy_to_client_enabled, "Client proxy enabled", "If true, let the device use the client's (e.g. your phone's) network connection to connect to the MQTT server. If false, it uses the device's network connection which you have to enable via the network settings."
map_reporting_enabled, "Map reporting enabled", "Available from firmware version 2.3.2 on. If true, your node will periodically send an unencrypted map report to the MQTT server to be displayed by online maps that support this packet."
map_report_settings, "Map report settings", "Settings for the map report module."
map_report_settings.publish_interval_secs, "Map report publish interval", "How often we should publish the map report to the MQTT server in seconds. Defaults to 900 seconds (15 minutes)."
map_report_settings.position_precision, "Map report position precision", "The precision to use for the position in the map report. Defaults to a maximum deviation of around 1459m."
map_report_settings.should_report_location, "Should report location", "Whether we have opted-in to report our location to the map."
[module.serial]
title, "Serial"
@@ -212,9 +208,9 @@ override_console_serial_port, "Override console serial port", "If set to true, t
title, "External Notification"
enabled, "Enabled", "Enables the module."
output_ms, "Length", "Specifies how long in milliseconds you would like your GPIOs to be active. In case of the repeat option, this is the duration of every tone and pause."
output, "Output GPIO", "Define the output pin GPIO setting Defaults to EXT_NOTIFY_OUT if set for the board. In standalone devices this pin should drive the LED to match the UI."
output_vibra, "Vibra GPIO", "Optional: Define a secondary output pin for a vibra motor. This is used in standalone devices to match the UI."
output_buzzer, "Buzzer GPIO", "Optional: Define a tertiary output pin for an active buzze. This is used in standalone devices to to match the UI."
output, "", ""
output_vibra, "", ""
output_buzzer, "", ""
active, "Active (general / LED only)", "Specifies whether the external circuit is active when the device's GPIO is low or high. If this is set true, the pin will be pulled active high, false means active low."
alert_message, "Alert when receiving a message (general)", "Specifies if an alert should be triggered when receiving an incoming message."
alert_message_vibra, "Alert vibration on message", "Specifies if a vibration alert should be triggered when receiving an incoming message."
@@ -284,8 +280,7 @@ i2s_sck, "I2S clock", "The GPIO to use for the SCK signal in the I2S interface."
[module.remote_hardware]
title, "Remote Hardware"
enabled, "Enabled", "Enables the module."
allow_undefined_pin_access, "Allow undefined pin access", "Whether the Module allows consumers to read / write to pins not defined in available_pins"
available_pins, "Available pins", "Exposes the available pins to the mesh for reading and writing."
allow_undefined_pin_access, "Allow undefined pin access", ""
[module.neighbor_info]
title, "Neighbor Info"
@@ -316,5 +311,5 @@ use_pullup, "Use pull-up", "Whether or not use INPUT_PULLUP mode for GPIO pin. O
title, "Paxcounter"
enabled, "Enabled", "Enables the module."
paxcounter_update_interval, "Update interval", "The interval in seconds of how often we can send a message to the mesh when a state change is detected."
Wi-Fi_threshold, "Wi-Fi Threshold", "WiFi RSSI threshold. Defaults to -80"
ble_threshold, "BLE Threshold", "BLE RSSI threshold. Defaults to -80"
Wi-Fi_threshold, "", ""
ble_threshold, "", ""

View File

@@ -2,46 +2,7 @@ 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 (
@@ -63,7 +24,7 @@ from contact.utilities.db_handler import (
)
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state, app_state, menu_state
from contact.utilities.singleton import ui_state, interface_state, app_state
def play_sound():
@@ -123,9 +84,6 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
if ui_state.display_log:
draw_packetlog_win()
if ui_state.current_window == 4:
menu_state.need_redraw = True
try:
if "decoded" not in packet:
return
@@ -140,14 +98,9 @@ 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":
schedule_notification_sound()
play_sound()
message_bytes = packet["decoded"]["payload"]
message_string = message_bytes.decode("utf-8")
@@ -184,7 +137,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} [{hops}] {message_from_string} ", message_string)
add_new_message(channel_id, f"{config.message_prefix} {message_from_string} ", message_string)
if refresh_channels:
draw_channel_list()

View File

@@ -1,5 +1,3 @@
import time
from typing import Any, Dict
import google.protobuf.json_format
@@ -51,7 +49,7 @@ def onAckNak(packet: Dict[str, Any]) -> None:
ack_type = "Nak"
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
config.sent_message_prefix + confirm_string + ": ",
message,
)

View File

@@ -1,108 +1,46 @@
import curses
import logging
import time
import traceback
from typing import Union
from contact.utilities.utils import get_channels, get_readable_duration, get_time_ago, refresh_node_list
from contact.settings import settings_menu
from contact.message_handlers.tx_handler import send_message, send_traceroute
from contact.utilities.utils import parse_protobuf
from contact.ui.colors import get_color
from contact.utilities.db_handler import get_name_from_database, update_node_info_in_db, is_chat_archived
from contact.utilities.input_handlers import get_list_input
import contact.ui.default_config as config
import contact.ui.dialog
from contact.ui.nav_utils import move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
from contact.utilities.singleton import ui_state, interface_state, menu_state
MIN_COL = 1 # "effectively zero" without breaking curses
root_win = None
# Draw arrows for a specific window id (0=channel,1=messages,2=nodes).
def draw_window_arrows(window_id: int) -> None:
if window_id == 0:
draw_main_arrows(channel_win, len(ui_state.channel_list), window=0)
channel_win.refresh()
elif window_id == 1:
msg_line_count = messages_pad.getmaxyx()[0]
draw_main_arrows(
messages_win,
msg_line_count,
window=1,
log_height=packetlog_win.getmaxyx()[0],
)
messages_win.refresh()
elif window_id == 2:
draw_main_arrows(nodes_win, len(ui_state.node_list), window=2)
nodes_win.refresh()
def compute_widths(total_w: int, focus: int):
# focus: 0=channel, 1=messages, 2=nodes
if total_w < 3 * MIN_COL:
# tiny terminals: allocate something, anything
return max(1, total_w), 0, 0
if focus == 0:
return total_w - 2 * MIN_COL, MIN_COL, MIN_COL
if focus == 1:
return MIN_COL, total_w - 2 * MIN_COL, MIN_COL
return MIN_COL, MIN_COL, total_w - 2 * MIN_COL
def paint_frame(win, selected: bool) -> None:
win.attrset(get_color("window_frame_selected") if selected else get_color("window_frame"))
win.box()
win.attrset(get_color("window_frame"))
win.refresh()
from contact.utilities.singleton import ui_state, interface_state
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, packetlog_win, entry_win
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, function_win, packetlog_win, entry_win
# Calculate window max dimensions
height, width = stdscr.getmaxyx()
if ui_state.single_pane_mode:
channel_width, messages_width, nodes_width = compute_widths(width, ui_state.current_window)
else:
channel_width = int(config.channel_list_16ths) * (width // 16)
nodes_width = int(config.node_list_16ths) * (width // 16)
messages_width = width - channel_width - nodes_width
channel_width = max(MIN_COL, channel_width)
messages_width = max(MIN_COL, messages_width)
nodes_width = max(MIN_COL, nodes_width)
# Ensure the three widths sum exactly to the terminal width by adjusting the focused pane
total = channel_width + messages_width + nodes_width
if total != width:
delta = total - width
if ui_state.current_window == 0:
channel_width = max(MIN_COL, channel_width - delta)
elif ui_state.current_window == 1:
messages_width = max(MIN_COL, messages_width - delta)
else:
nodes_width = max(MIN_COL, nodes_width - delta)
# Define window dimensions and positions
channel_width = int(config.channel_list_16ths) * (width // 16)
nodes_width = int(config.node_list_16ths) * (width // 16)
messages_width = width - channel_width - nodes_width
entry_height = 3
y_pad = entry_height
content_h = max(1, height - y_pad)
pkt_h = max(1, int(height / 3))
function_height = 3
y_pad = entry_height + function_height
packet_log_height = int(height / 3)
if firstrun:
entry_win = curses.newwin(entry_height, width, height - entry_height, 0)
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)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, channel_width)
entry_win = curses.newwin(entry_height, width, 0, 0)
channel_win = curses.newwin(height - y_pad, channel_width, entry_height, 0)
messages_win = curses.newwin(height - y_pad, messages_width, entry_height, channel_width)
nodes_win = curses.newwin(height - y_pad, nodes_width, entry_height, channel_width + messages_width)
function_win = curses.newwin(function_height, width, height - function_height, 0)
packetlog_win = curses.newwin(
packet_log_height, messages_width, height - packet_log_height - function_height, channel_width
)
# Will be resized to what we need when drawn
messages_pad = curses.newpad(1, 1)
@@ -110,7 +48,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, packetlog_win]:
for win in [entry_win, channel_win, messages_win, nodes_win, function_win, packetlog_win]:
win.bkgd(get_color("background"))
# Set background colors for pads
@@ -118,30 +56,31 @@ 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]:
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
win.attrset(get_color("window_frame"))
else:
for win in [entry_win, channel_win, messages_win, nodes_win, packetlog_win]:
for win in [entry_win, channel_win, messages_win, nodes_win, function_win, packetlog_win]:
win.erase()
entry_win.resize(entry_height, width)
entry_win.mvwin(height - entry_height, 0)
entry_win.resize(3, width)
channel_win.resize(content_h, channel_width)
channel_win.mvwin(0, 0)
channel_win.resize(height - y_pad, channel_width)
messages_win.resize(content_h, messages_width)
messages_win.mvwin(0, channel_width)
messages_win.resize(height - y_pad, messages_width)
messages_win.mvwin(3, channel_width)
nodes_win.resize(content_h, nodes_width)
nodes_win.mvwin(0, channel_width + messages_width)
nodes_win.resize(height - y_pad, nodes_width)
nodes_win.mvwin(entry_height, channel_width + messages_width)
packetlog_win.resize(pkt_h, messages_width)
packetlog_win.mvwin(height - pkt_h - entry_height, channel_width)
function_win.resize(3, width)
function_win.mvwin(height - function_height, 0)
packetlog_win.resize(packet_log_height, messages_width)
packetlog_win.mvwin(height - packet_log_height - function_height, channel_width)
# Draw window borders
for win in [channel_win, entry_win, nodes_win, messages_win]:
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
win.box()
win.refresh()
@@ -149,10 +88,10 @@ 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()
draw_window_arrows(ui_state.current_window)
except:
# Resize events can come faster than we can re-draw, which can cause a curses error.
@@ -163,16 +102,13 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
def main_ui(stdscr: curses.window) -> None:
"""Main UI loop for the curses interface."""
global input_text
global root_win
root_win = stdscr
input_text = ""
stdscr.keypad(True)
get_channels()
handle_resize(stdscr, True)
while True:
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
draw_text_field(entry_win, f"Input: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
# Get user input from entry window
char = entry_win.get_wch()
@@ -200,22 +136,16 @@ 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 in (curses.KEY_F4, chr(20)): # Ctrl + t and F4 for Traceroute
elif char == chr(20): # Ctrl + t 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 in (curses.KEY_F12, "`"): # ` Launch the settings interface
elif char == "`": # ` Launch the settings interface
handle_backtick(stdscr)
elif char == chr(16): # Ctrl + P for Packet Log
@@ -231,9 +161,6 @@ 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)
@@ -281,8 +208,6 @@ def handle_home() -> None:
elif ui_state.current_window == 2:
select_node(0)
draw_window_arrows(ui_state.current_window)
def handle_end() -> None:
"""Handle end key events to select the last item in the current window."""
@@ -294,27 +219,29 @@ def handle_end() -> None:
refresh_pad(1)
elif ui_state.current_window == 2:
select_node(len(ui_state.node_list) - 1)
draw_window_arrows(ui_state.current_window)
def handle_pageup() -> None:
"""Handle page up key events to scroll the current window by a page."""
if ui_state.current_window == 0:
select_channel(ui_state.selected_channel - (channel_win.getmaxyx()[0] - 2))
select_channel(
ui_state.selected_channel - (channel_win.getmaxyx()[0] - 2)
) # select_channel will bounds check for us
elif ui_state.current_window == 1:
ui_state.selected_message = max(
ui_state.selected_message - get_msg_window_lines(messages_win, packetlog_win), 0
)
refresh_pad(1)
elif ui_state.current_window == 2:
select_node(ui_state.selected_node - (nodes_win.getmaxyx()[0] - 2))
draw_window_arrows(ui_state.current_window)
select_node(ui_state.selected_node - (nodes_win.getmaxyx()[0] - 2)) # select_node will bounds check for us
def handle_pagedown() -> None:
"""Handle page down key events to scroll the current window down."""
if ui_state.current_window == 0:
select_channel(ui_state.selected_channel + (channel_win.getmaxyx()[0] - 2))
select_channel(
ui_state.selected_channel + (channel_win.getmaxyx()[0] - 2)
) # select_channel will bounds check for us
elif ui_state.current_window == 1:
msg_line_count = messages_pad.getmaxyx()[0]
ui_state.selected_message = min(
@@ -323,8 +250,7 @@ def handle_pagedown() -> None:
)
refresh_pad(1)
elif ui_state.current_window == 2:
select_node(ui_state.selected_node + (nodes_win.getmaxyx()[0] - 2))
draw_window_arrows(ui_state.current_window)
select_node(ui_state.selected_node + (nodes_win.getmaxyx()[0] - 2)) # select_node will bounds check for us
def handle_leftright(char: int) -> None:
@@ -332,78 +258,44 @@ def handle_leftright(char: int) -> None:
delta = -1 if char == curses.KEY_LEFT else 1
old_window = ui_state.current_window
ui_state.current_window = (ui_state.current_window + delta) % 3
handle_resize(root_win, False)
if old_window == 0:
paint_frame(channel_win, selected=False)
channel_win.attrset(get_color("window_frame"))
channel_win.box()
channel_win.refresh()
refresh_pad(0)
if old_window == 1:
paint_frame(messages_win, selected=False)
messages_win.attrset(get_color("window_frame"))
messages_win.box()
messages_win.refresh()
refresh_pad(1)
elif old_window == 2:
paint_frame(nodes_win, selected=False)
draw_function_win()
nodes_win.attrset(get_color("window_frame"))
nodes_win.box()
nodes_win.refresh()
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)
channel_win.attrset(get_color("window_frame_selected"))
channel_win.box()
channel_win.attrset(get_color("window_frame"))
channel_win.refresh()
refresh_pad(0)
elif ui_state.current_window == 1:
paint_frame(messages_win, selected=True)
messages_win.attrset(get_color("window_frame_selected"))
messages_win.box()
messages_win.attrset(get_color("window_frame"))
messages_win.refresh()
refresh_pad(1)
elif ui_state.current_window == 2:
paint_frame(nodes_win, selected=True)
draw_function_win()
nodes_win.attrset(get_color("window_frame_selected"))
nodes_win.box()
nodes_win.attrset(get_color("window_frame"))
nodes_win.refresh()
refresh_pad(2)
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)
def handle_enter(input_text: str) -> str:
"""Handle Enter key events to send messages or select channels."""
@@ -422,138 +314,29 @@ def handle_enter(input_text: str) -> str:
ui_state.selected_node = 0
ui_state.current_window = 0
handle_resize(root_win, False)
draw_node_list()
draw_channel_list()
draw_messages_window(True)
draw_window_arrows(ui_state.current_window)
return input_text
elif len(input_text) > 0:
# TODO: This is a hack to prevent sending messages too quickly. Let's get errors from the node.
now = time.monotonic()
if now - ui_state.last_sent_time < 2.5:
contact.ui.dialog.dialog("Slow down", "Please wait 2 seconds between messages.")
return input_text
# Enter key pressed, send user input as message
send_message(input_text, channel=ui_state.selected_channel)
draw_messages_window(True)
ui_state.last_sent_time = now
# Clear entry window and reset input text
entry_win.erase()
if ui_state.current_window == 0:
ui_state.current_window = 1
handle_resize(root_win, False)
return ""
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.",
"Results will appear in messages window.\nNote: Traceroute is limited to once every 30 seconds.",
)
curses.curs_set(1) # Show cursor again
handle_resize(stdscr, False)
@@ -574,10 +357,7 @@ def handle_backspace(entry_win: curses.window, input_text: str) -> str:
def handle_backtick(stdscr: curses.window) -> None:
"""Handle backtick key events to open the settings menu."""
curses.curs_set(0)
previous_window = ui_state.current_window
ui_state.current_window = 4
settings_menu(stdscr, interface_state.interface)
ui_state.current_window = previous_window
curses.curs_set(1)
refresh_node_list()
handle_resize(stdscr, False)
@@ -595,34 +375,6 @@ 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):
@@ -738,10 +490,6 @@ def handle_ctlr_g(stdscr: curses.window) -> None:
def draw_channel_list() -> None:
"""Update the channel list window and pad based on the current state."""
if ui_state.current_window != 0 and ui_state.single_pane_mode:
return
channel_pad.erase()
win_width = channel_win.getmaxyx()[1]
@@ -776,18 +524,20 @@ def draw_channel_list() -> None:
channel_pad.addstr(idx, 1, truncated_channel, color)
idx += 1
paint_frame(channel_win, selected=(ui_state.current_window == 0))
refresh_pad(0)
draw_window_arrows(0)
channel_win.attrset(
get_color("window_frame_selected") if ui_state.current_window == 0 else get_color("window_frame")
)
channel_win.box()
channel_win.attrset((get_color("window_frame")))
draw_main_arrows(channel_win, len(ui_state.channel_list), window=0)
channel_win.refresh()
refresh_pad(0)
def draw_messages_window(scroll_to_bottom: bool = False) -> None:
"""Update the messages window based on the selected channel and scroll position."""
if ui_state.current_window != 1 and ui_state.single_pane_mode:
return
messages_pad.erase()
channel = ui_state.channel_list[ui_state.selected_channel]
@@ -807,7 +557,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.find(config.sent_message_prefix) != -1:
elif prefix.startswith(config.sent_message_prefix):
color = get_color("tx_messages")
else:
color = get_color("rx_messages")
@@ -815,32 +565,39 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
messages_pad.addstr(row, 1, line, color)
row += 1
paint_frame(messages_win, selected=(ui_state.current_window == 1))
messages_win.attrset(
get_color("window_frame_selected") if ui_state.current_window == 1 else get_color("window_frame")
)
messages_win.box()
messages_win.attrset(get_color("window_frame"))
messages_win.refresh()
visible_lines = get_msg_window_lines(messages_win, packetlog_win)
if scroll_to_bottom:
ui_state.selected_message = max(msg_line_count - visible_lines, 0)
ui_state.start_index[1] = max(msg_line_count - visible_lines, 0)
pass
else:
ui_state.selected_message = max(min(ui_state.selected_message, msg_line_count - visible_lines), 0)
draw_main_arrows(
messages_win,
msg_line_count,
window=1,
log_height=packetlog_win.getmaxyx()[0],
)
messages_win.refresh()
refresh_pad(1)
draw_packetlog_win()
draw_window_arrows(1)
messages_win.refresh()
if ui_state.current_window == 4:
menu_state.need_redraw = True
def draw_node_list() -> None:
"""Update the nodes list window and pad based on the current state."""
global nodes_pad
if ui_state.current_window != 2 and ui_state.single_pane_mode:
return
# This didn't work, for some reason an error is thown on startup, so we just create the pad every time
# if nodes_pad is None:
# nodes_pad = curses.newpad(1, 1)
@@ -857,21 +614,9 @@ 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"]
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]
node_str = f"{'🔐' if secure else '🔓'} {get_name_from_database(node_num, 'long')}".ljust(box_width - 2)[
: box_width - 2
]
color = "node_list"
if "isFavorite" in node and node["isFavorite"]:
color = "node_favorite"
@@ -881,20 +626,22 @@ def draw_node_list() -> None:
i, 1, node_str, get_color(color, reverse=ui_state.selected_node == i and ui_state.current_window == 2)
)
paint_frame(nodes_win, selected=(ui_state.current_window == 2))
nodes_win.attrset(
get_color("window_frame_selected") if ui_state.current_window == 2 else get_color("window_frame")
)
nodes_win.box()
nodes_win.attrset(get_color("window_frame"))
draw_main_arrows(nodes_win, len(ui_state.node_list), window=2)
nodes_win.refresh()
refresh_pad(2)
draw_window_arrows(2)
nodes_win.refresh()
# Restore cursor to input field
entry_win.keypad(True)
curses.curs_set(1)
entry_win.refresh()
if ui_state.current_window == 4:
menu_state.need_redraw = True
def select_channel(idx: int) -> None:
"""Select a channel by index and update the UI state accordingly."""
@@ -952,9 +699,15 @@ def scroll_messages(direction: int) -> None:
0, min(ui_state.start_index[ui_state.current_window], max_index - visible_height + 1)
)
draw_main_arrows(
messages_win,
msg_line_count,
ui_state.current_window,
log_height=packetlog_win.getmaxyx()[0],
)
messages_win.refresh()
refresh_pad(1)
draw_window_arrows(ui_state.current_window)
def select_node(idx: int) -> None:
@@ -971,6 +724,8 @@ 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."""
@@ -989,9 +744,6 @@ def draw_packetlog_win() -> None:
columns = [10, 10, 15, 30]
span = 0
if ui_state.current_window != 1 and ui_state.single_pane_mode:
return
if ui_state.display_log:
packetlog_win.erase()
height, width = packetlog_win.getmaxyx()
@@ -1017,20 +769,22 @@ def draw_packetlog_win() -> None:
else get_name_from_database(packet["to"], "short").ljust(columns[1])
)
if "decoded" in packet:
port = str(packet["decoded"].get("portnum", "")).ljust(columns[2])
parsed_payload = parse_protobuf(packet)
port = packet["decoded"]["portnum"].ljust(columns[2])
payload = (packet["decoded"]["payload"]).ljust(columns[3])
else:
port = "NO KEY".ljust(columns[2])
parsed_payload = "NO KEY"
payload = "NO KEY".ljust(columns[3])
# Combine and truncate if necessary
logString = f"{from_id} {to_id} {port} {parsed_payload}"
logString = f"{from_id} {to_id} {port} {payload}"
logString = logString[: width - 3]
# Add to the window
packetlog_win.addstr(i + 2, 1, logString, get_color("log"))
paint_frame(packetlog_win, selected=False)
packetlog_win.attrset(get_color("window_frame"))
packetlog_win.box()
packetlog_win.refresh()
# Restore cursor to input field
entry_win.keypad(True)
@@ -1087,13 +841,101 @@ def search(win: int) -> None:
entry_win.erase()
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:
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:
return
# Derive the target box and pad for the requested window
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:
win_height = channel_win.getmaxyx()[0]
if window == 1:
@@ -1111,7 +953,6 @@ 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
@@ -1122,39 +963,13 @@ 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
# Compute inner drawable area of the box
box_y, box_x = box.getbegyx()
box_h, box_w = box.getmaxyx()
inner_h = max(0, box_h - 2) # minus borders
inner_w = max(0, box_w - 2)
if inner_h <= 0 or inner_w <= 0:
return
# Clamp lines to available inner height
lines = max(0, min(lines, inner_h))
# Clamp start_index within the pad's height
pad_h, pad_w = pad.getmaxyx()
if pad_h <= 0:
return
start_index = max(0, min(start_index, max(0, pad_h - 1)))
top = box_y + 1
left = box_x + 1
bottom = box_y + min(inner_h, lines) # inclusive
right = box_x + min(inner_w, box_w - 2)
if bottom < top or right < left:
return
pad.refresh(
start_index,
0,
top,
left,
bottom,
right,
box.getbegyx()[0] + 1,
box.getbegyx()[1] + 1,
box.getbegyx()[0] + lines,
box.getbegyx()[1] + box.getmaxyx()[1] - 3,
)
@@ -1170,23 +985,7 @@ def remove_notification(channel_number: int) -> None:
def draw_text_field(win: curses.window, text: str, color: int) -> None:
win.border()
# 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
win.addstr(1, 1, text, color)
def draw_centered_text_field(win: curses.window, text: str, y_offset: int, color: int) -> None:
@@ -1195,3 +994,8 @@ 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()

View File

@@ -6,7 +6,6 @@ import sys
from typing import List
from contact.utilities.save_to_radio import save_changes
import contact.ui.default_config as config
from contact.utilities.config_io import config_export, config_import
from contact.utilities.control_utils import parse_ini_file, transform_menu_path
from contact.utilities.input_handlers import (
@@ -21,74 +20,60 @@ from contact.ui.dialog import dialog
from contact.ui.menus import generate_menu_from_protobuf
from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
from contact.ui.user_config import json_editor
from contact.utilities.singleton import menu_state
from contact.ui.ui_state import MenuState
# Setup Variables
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
menu_state = MenuState()
# Constants
width = 80
save_option = "Save Changes"
max_help_lines = 0
help_win = None
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
# Compute the effective menu width for the current terminal
def get_menu_width() -> int:
# Leave at least 2 columns for borders; clamp to >= 20 for usability
return max(20, min(MAX_MENU_WIDTH, curses.COLS - 2))
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
# Get the parent directory of the script
script_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.abspath(os.path.join(script_dir, os.pardir))
# Paths
# locals_dir = os.path.dirname(os.path.abspath(sys.argv[0])) # Current script directory
locals_dir = os.path.dirname(os.path.abspath(sys.argv[0])) # Current script directory
translation_file = os.path.join(parent_dir, "localisations", "en.ini")
# config_folder = os.path.join(locals_dir, "node-configs")
config_folder = os.path.abspath(config.node_configs_file_path)
config_folder = os.path.join(locals_dir, "node-configs")
# Load translations
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
def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.window or pad types
min_help_window_height = 6
num_items = len(menu_state.current_menu) + (1 if menu_state.show_save_option else 0)
# Determine the available height for the menu
max_menu_height = curses.LINES
menu_height = min(max_menu_height - min_help_window_height, num_items + 5)
w = get_menu_width()
start_y = (curses.LINES - menu_height) // 2 - (min_help_window_height // 2)
start_x = (curses.COLS - w) // 2
start_x = (curses.COLS - width) // 2
# Calculate remaining space for help window
global max_help_lines
remaining_space = curses.LINES - (start_y + menu_height + 2) # +2 for padding
max_help_lines = max(remaining_space, 1) # Ensure at least 1 lines for help
menu_win = curses.newwin(menu_height, w, start_y, start_x)
menu_win = curses.newwin(menu_height, width, start_y, start_x)
menu_win.erase()
menu_win.bkgd(get_color("background"))
menu_win.attrset(get_color("window_frame"))
menu_win.border()
menu_win.keypad(True)
menu_pad = curses.newpad(len(menu_state.current_menu) + 1, w - 8)
menu_pad = curses.newpad(len(menu_state.current_menu) + 1, width - 8)
menu_pad.bkgd(get_color("background"))
header = " > ".join(word.title() for word in menu_state.menu_path)
if len(header) > w - 4:
header = header[: w - 7] + "..."
if len(header) > width - 4:
header = header[: width - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
transformed_path = transform_menu_path(menu_state.menu_path)
@@ -99,15 +84,15 @@ def display_menu() -> tuple[object, object]:
full_key = ".".join(transformed_path + [option])
display_name = field_mapping.get(full_key, option)
display_option = f"{display_name}"[: w // 2 - 2]
display_value = f"{current_value}"[: w // 2 - 4]
display_option = f"{display_name}"[: width // 2 - 2]
display_value = f"{current_value}"[: width // 2 - 4]
try:
color = get_color(
"settings_sensitive" if option in sensitive_settings else "settings_default",
reverse=(idx == menu_state.selected_index),
)
menu_pad.addstr(idx, 0, f"{display_option:<{w // 2 - 2}} {display_value}".ljust(w - 8), color)
menu_pad.addstr(idx, 0, f"{display_option:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
except curses.error:
pass
@@ -115,13 +100,13 @@ def display_menu() -> tuple[object, object]:
save_position = menu_height - 2
menu_win.addstr(
save_position,
(w - len(save_option)) // 2,
(width - len(save_option)) // 2,
save_option,
get_color("settings_save", reverse=(menu_state.selected_index == len(menu_state.current_menu))),
)
# Draw help window with dynamically updated max_help_lines
draw_help_window(start_y, start_x, menu_height, max_help_lines, transformed_path)
draw_help_window(start_y, start_x, menu_height, max_help_lines, transformed_path, menu_state)
menu_win.refresh()
menu_pad.refresh(
@@ -132,7 +117,6 @@ def display_menu() -> tuple[object, object]:
menu_win.getbegyx()[0] + 3 + menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0),
menu_win.getbegyx()[1] + menu_win.getmaxyx()[1] - 4,
)
curses.curs_set(0)
max_index = num_items + (1 if menu_state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0)
@@ -148,7 +132,9 @@ def draw_help_window(
menu_height: int,
max_help_lines: int,
transformed_path: List[str],
menu_state: MenuState,
) -> None:
global help_win
if "help_win" not in globals():
@@ -159,9 +145,8 @@ def draw_help_window(
)
help_y = menu_start_y + menu_height
# Use current terminal width for the help window width calculation
help_win = update_help_window(
help_win, help_text, transformed_path, selected_option, max_help_lines, get_menu_width(), help_y, menu_start_x
help_win, help_text, transformed_path, selected_option, max_help_lines, width, help_y, menu_start_x
)
@@ -183,33 +168,29 @@ def settings_menu(stdscr: object, interface: object) -> None:
modified_settings = {}
menu_state.need_redraw = True
need_redraw = True
menu_state.show_save_option = False
new_value_name = None
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
if need_redraw:
options = list(menu_state.current_menu.keys())
# Determine if save option should be shown
path = menu_state.menu_path
menu_state.show_save_option = (
(len(path) > 2 and ("Radio Settings" in path or "Module Settings" in path))
or (len(path) == 2 and "User Settings" in path)
or (len(path) == 3 and "Channels" in path)
(
len(menu_state.menu_path) > 2
and ("Radio Settings" in menu_state.menu_path or "Module Settings" in menu_state.menu_path)
)
or (len(menu_state.menu_path) == 2 and "User Settings" in menu_state.menu_path)
or (len(menu_state.menu_path) == 3 and "Channels" in menu_state.menu_path)
)
# Display the menu
menu_win, menu_pad = display_menu()
menu_win, menu_pad = display_menu(menu_state)
if menu_win is None:
continue # Skip if menu_win is not initialized
need_redraw = False
menu_win.timeout(200) # wait up to 200 ms for a keypress (or less if key is pressed)
# Capture user input
key = menu_win.getch()
if key == -1:
continue
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
# max_help_lines = 4
@@ -243,16 +224,14 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
elif key == curses.KEY_RESIZE:
menu_state.need_redraw = True
need_redraw = True
curses.update_lines_cols()
menu_win.erase()
if help_win:
help_win.erase()
help_win.erase()
menu_win.refresh()
if help_win:
help_win.refresh()
help_win.refresh()
elif key == ord("\t") and menu_state.show_save_option:
old_selected_index = menu_state.selected_index
@@ -269,17 +248,15 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
elif key == curses.KEY_RIGHT or key == ord("\n"):
menu_state.need_redraw = True
need_redraw = True
menu_state.start_index.append(0)
menu_win.erase()
if help_win:
help_win.erase()
help_win.erase()
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, menu_state.current_menu, selected_index, transform_menu_path(menu_state.menu_path))
menu_win.refresh()
if help_win:
help_win.refresh()
help_win.refresh()
if menu_state.show_save_option and menu_state.selected_index == len(options):
save_changes(interface, modified_settings, menu_state)
@@ -324,7 +301,6 @@ def settings_menu(stdscr: object, interface: object) -> None:
file.write(config_text)
logging.info(f"Config file saved to {yaml_file_path}")
dialog("Config File Saved:", yaml_file_path)
menu_state.need_redraw = True
menu_state.start_index.pop()
continue
except PermissionError:
@@ -341,7 +317,6 @@ def settings_menu(stdscr: object, interface: object) -> None:
# Check if folder exists and is not empty
if not os.path.exists(config_folder) or not any(os.listdir(config_folder)):
dialog("", " No config files found. Export a config first.")
menu_state.need_redraw = True
continue # Return to menu
file_list = [f for f in os.listdir(config_folder) if os.path.isfile(os.path.join(config_folder, f))]
@@ -349,7 +324,6 @@ def settings_menu(stdscr: object, interface: object) -> None:
# Ensure file_list is not empty before proceeding
if not file_list:
dialog("", " No config files found. Export a config first.")
menu_state.need_redraw = True
continue
filename = get_list_input("Choose a config file", None, file_list)
@@ -416,6 +390,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
menu_state.start_index.pop()
menu_state.selected_index = 4
continue
# need_redraw = True
field_info = menu_state.current_menu.get(selected_option)
if isinstance(field_info, tuple):
@@ -518,16 +493,8 @@ def settings_menu(stdscr: object, interface: object) -> None:
for key in menu_state.menu_path[3:]: # Skip "Main Menu"
modified_settings = modified_settings.setdefault(key, {})
# For comparison, normalize enum numbers to names
compare_value = new_value
if field and field.enum_type and isinstance(new_value, int):
enum_value_descriptor = field.enum_type.values_by_number.get(new_value)
if enum_value_descriptor:
compare_value = enum_value_descriptor.name
if compare_value != current_value:
# Save the raw protobuf number, not the name
modified_settings[selected_option] = new_value
# Add the new value to the appropriate level
modified_settings[selected_option] = new_value
# Convert enum string to int
if field and field.enum_type:
@@ -542,42 +509,19 @@ def settings_menu(stdscr: object, interface: object) -> None:
menu_state.selected_index = 0
elif key == curses.KEY_LEFT:
# If we are at the main menu and there are unsaved changes, prompt to save
if len(menu_state.menu_path) == 3 and modified_settings:
current_section = menu_state.menu_path[-1]
save_prompt = get_list_input(
f"You have unsaved changes in {current_section}. Save before exiting?",
None,
["Yes", "No", "Cancel"],
mandatory=True,
)
if save_prompt == "Cancel":
continue # Stay in the menu without doing anything
elif save_prompt == "Yes":
save_changes(interface, modified_settings, menu_state)
logging.info("Changes Saved")
modified_settings.clear()
menu = rebuild_menu_at_current_path(interface, menu_state)
pass
menu_state.need_redraw = True
need_redraw = True
menu_win.erase()
if help_win:
help_win.erase()
help_win.erase()
# max_help_lines = 4
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, menu_state.current_menu, selected_index, transform_menu_path(menu_state.menu_path))
menu_win.refresh()
if help_win:
help_win.refresh()
help_win.refresh()
# if len(menu_state.menu_path) < 2:
# modified_settings.clear()
if len(menu_state.menu_path) < 2:
modified_settings.clear()
# Navigate back to the previous menu
if len(menu_state.menu_path) > 1:
@@ -594,16 +538,6 @@ def settings_menu(stdscr: object, interface: object) -> None:
break
def rebuild_menu_at_current_path(interface, menu_state):
"""Rebuild menus from the device and re-point current_menu to the same path."""
new_menu = generate_menu_from_protobuf(interface)
cur = new_menu["Main Menu"]
for step in menu_state.menu_path[1:]:
cur = cur.get(step, {})
menu_state.current_menu = cur
return new_menu
def set_region(interface: object) -> None:
node = interface.getNode("^local")
device_config = node.localConfig

View File

@@ -2,7 +2,6 @@ import json
import logging
import os
from typing import Dict
from contact.ui.colors import setup_colors
# Get the parent directory of the script
script_dir = os.path.dirname(os.path.abspath(__file__))
@@ -14,12 +13,6 @@ parent_dir = os.path.abspath(os.path.join(script_dir, os.pardir))
# parent_dir = "/tmp/test_nonwritable"
def reload_config() -> None:
loaded_config = initialize_config()
assign_config_variables(loaded_config)
setup_colors(reinit=True)
def _is_writable_dir(path: str) -> bool:
"""
Return True if we can create & delete a temp file in `path`.
@@ -64,7 +57,6 @@ config_root = _get_config_root(parent_dir)
json_file_path = os.path.join(config_root, "config.json")
log_file_path = os.path.join(config_root, "client.log")
db_file_path = os.path.join(config_root, "client.db")
node_configs_file_path = os.path.join(config_root, "node-configs/")
def format_json_single_line_arrays(data: Dict[str, object], indent: int = 4) -> str:
@@ -183,10 +175,8 @@ def initialize_config() -> Dict[str, object]:
default_config_variables = {
"channel_list_16ths": "3",
"node_list_16ths": "5",
"single_pane_mode": "False",
"db_file_path": db_file_path,
"log_file_path": log_file_path,
"node_configs_file_path": node_configs_file_path,
"message_prefix": ">>",
"sent_message_prefix": ">> Sent",
"notification_symbol": "*",
@@ -227,18 +217,16 @@ def initialize_config() -> Dict[str, object]:
def assign_config_variables(loaded_config: Dict[str, object]) -> None:
# Assign values to local variables
global db_file_path, log_file_path, node_configs_file_path, message_prefix, sent_message_prefix
global db_file_path, log_file_path, message_prefix, sent_message_prefix
global notification_symbol, ack_implicit_str, ack_str, nak_str, ack_unknown_str
global node_list_16ths, channel_list_16ths, single_pane_mode
global node_list_16ths, channel_list_16ths
global theme, COLOR_CONFIG
global node_sort, notification_sound
channel_list_16ths = loaded_config["channel_list_16ths"]
node_list_16ths = loaded_config["node_list_16ths"]
single_pane_mode = loaded_config["single_pane_mode"]
db_file_path = loaded_config["db_file_path"]
log_file_path = loaded_config["log_file_path"]
node_configs_file_path = loaded_config.get("node_configs_file_path")
message_prefix = loaded_config["message_prefix"]
sent_message_prefix = loaded_config["sent_message_prefix"]
notification_symbol = loaded_config["notification_symbol"]
@@ -270,7 +258,6 @@ if __name__ == "__main__":
print("\nLoaded Configuration:")
print(f"Database File Path: {db_file_path}")
print(f"Log File Path: {log_file_path}")
print(f"Configs File Path: {node_configs_file_path}")
print(f"Message Prefix: {message_prefix}")
print(f"Sent Message Prefix: {sent_message_prefix}")
print(f"Notification Symbol: {notification_symbol}")

View File

@@ -1,164 +1,49 @@
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
def dialog(title: str, message: str) -> None:
"""Display a dialog with a title and message."""
previous_window = ui_state.current_window
ui_state.current_window = 4
curses.update_lines_cols()
height, width = curses.LINES, curses.COLS
# Parse message into lines and calculate dimensions
message_lines = message.splitlines() or [""]
# Calculate dialog dimensions
message_lines = message.splitlines()
max_line_length = max(len(l) for l in message_lines)
# Desired size
dialog_width = max(len(title) + 4, max_line_length + 4)
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()
win.bkgd(get_color("background"))
win.attrset(get_color("window_frame"))
win.border(0)
# Title
try:
win.addstr(0, 2, title[: max(0, dialog_width - 4)], get_color("settings_default"))
except curses.error:
pass
# 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 "
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.noutrefresh()
curses.doupdate()
dialog_height = len(message_lines) + 4
x = (width - dialog_width) // 2
y = (height - dialog_height) // 2
# Create dialog window
win = curses.newwin(dialog_height, dialog_width, y, x)
win.keypad(True)
draw_window()
win.bkgd(get_color("background"))
win.attrset(get_color("window_frame"))
win.border(0)
# Add title
win.addstr(0, 2, title, get_color("settings_default"))
# Add message (centered)
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"))
# Add centered OK button
ok_text = " Ok "
win.addstr(
dialog_height - 2,
(dialog_width - len(ok_text)) // 2,
ok_text,
get_color("settings_default", reverse=True),
)
# Refresh dialog window
win.refresh()
# Get user input
while True:
win.timeout(200)
char = win.getch()
if menu_state.need_redraw:
menu_state.need_redraw = False
curses.update_lines_cols()
height, width = curses.LINES, curses.COLS
draw_window()
# Close dialog
ok_selected = True
if char in (27, curses.KEY_LEFT): # Esc or Left arrow
if char in (curses.KEY_ENTER, 10, 13, 32, 27): # Enter, space, or Esc
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
return
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()

View File

@@ -22,9 +22,9 @@ def get_node_color(node_index: int, reverse: bool = False):
Segment = tuple[str, str, bool, bool]
WrappedLine = List[Segment]
width = 80
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
save_option = "Save Changes"
MIN_HEIGHT_FOR_HELP = 20
def move_highlight(
@@ -73,8 +73,9 @@ def move_highlight(
# Clear old selection
if show_save_option and old_idx == max_index:
win_h, win_w = menu_win.getmaxyx()
menu_win.chgat(win_h - 2, (win_w - len(save_option)) // 2, len(save_option), get_color("settings_save"))
menu_win.chgat(
menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save")
)
else:
menu_pad.chgat(
old_idx,
@@ -89,10 +90,9 @@ def move_highlight(
# Highlight new selection
if show_save_option and new_idx == max_index:
win_h, win_w = menu_win.getmaxyx()
menu_win.chgat(
win_h - 2,
(win_w - len(save_option)) // 2,
menu_win.getmaxyx()[0] - 2,
(width - len(save_option)) // 2,
len(save_option),
get_color("settings_save", reverse=True),
)
@@ -124,14 +124,13 @@ def move_highlight(
selected_option = options[new_idx] if new_idx < len(options) else None
help_y = menu_win.getbegyx()[0] + menu_win.getmaxyx()[0]
if help_win:
win_h, win_w = menu_win.getmaxyx()
help_win = update_help_window(
help_win,
help_text,
transformed_path,
selected_option,
max_help_lines,
win_w,
width,
help_y,
menu_win.getbegyx()[1],
)
@@ -168,46 +167,23 @@ def update_help_window(
help_x: int,
) -> object: # returns a curses window
"""Handles rendering the help window consistently."""
if curses.LINES < MIN_HEIGHT_FOR_HELP:
return None
# Clamp target position and width to the current terminal size
help_x = max(0, help_x)
help_y = max(0, help_y)
# Ensure requested width fits on screen from help_x
max_w_from_x = max(1, curses.COLS - help_x)
safe_width = min(width, max_w_from_x)
# Always leave a minimal border area; enforce a minimum usable width of 3
safe_width = max(3, safe_width)
wrapped_help = get_wrapped_help_text(help_text, transformed_path, selected_option, safe_width, max_help_lines)
wrapped_help = get_wrapped_help_text(help_text, transformed_path, selected_option, width, max_help_lines)
help_height = min(len(wrapped_help) + 2, max_help_lines + 2) # +2 for border
help_height = max(help_height, 3) # Ensure at least 3 rows (1 text + border)
# Re-clamp Y to keep the window visible
# Ensure help window does not exceed screen size
if help_y + help_height > curses.LINES:
help_y = max(0, curses.LINES - help_height)
# If width would overflow the screen, shrink it
if help_x + safe_width > curses.COLS:
safe_width = max(3, curses.COLS - help_x)
help_y = curses.LINES - help_height
# Create or update the help window
if help_win is None:
help_win = curses.newwin(help_height, safe_width, help_y, help_x)
help_win = curses.newwin(help_height, width, help_y, help_x)
else:
help_win.erase()
help_win.refresh()
help_win.resize(help_height, safe_width)
try:
help_win.mvwin(help_y, help_x)
except curses.error:
# If moving fails due to edge conditions, pin to (0,0) as a fallback
help_y = 0
help_x = 0
help_win.mvwin(help_y, help_x)
help_win.resize(help_height, width)
help_win.mvwin(help_y, help_x)
help_win.bkgd(get_color("background"))
help_win.attrset(get_color("window_frame"))
@@ -319,16 +295,14 @@ def get_wrapped_help_text(
return wrapped_help
def text_width(text: str) -> int:
return sum(2 if east_asian_width(c) in "FW" else 1 for c in text)
def wrap_text(text: str, wrap_width: int) -> List[str]:
"""Wraps text while preserving spaces and breaking long words."""
whitespace = "\t\n\x0b\x0c\r "
whitespace_trans = dict.fromkeys(map(ord, whitespace), ord(" "))
whitespace = '\t\n\x0b\x0c\r '
whitespace_trans = dict.fromkeys(map(ord, whitespace), ord(' '))
text = text.translate(whitespace_trans)
words = re.findall(r"\S+|\s+", text) # Capture words and spaces separately

View File

@@ -22,7 +22,6 @@ def draw_splash(stdscr: object) -> None:
stdscr.addstr(start_y + 1, start_x - 1, message_2, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 2, start_x - 2, message_3, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 4, start_x2, message_4, get_color("splash_text"))
stdscr.move(start_y + 5, start_x2)
stdscr.attrset(get_color("window_frame"))
stdscr.box()

View File

@@ -10,7 +10,6 @@ class MenuState:
current_menu: Union[Dict[str, Any], List[Any], str, int] = field(default_factory=dict)
menu_path: List[str] = field(default_factory=list)
show_save_option: bool = False
need_redraw: bool = False
@dataclass
@@ -25,14 +24,11 @@ class ChatUIState:
selected_message: int = 0
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])
show_save_option: bool = False
menu_path: List[str] = field(default_factory=list)
single_pane_mode: bool = False
@dataclass

View File

@@ -4,23 +4,16 @@ import curses
from typing import Any, List, Dict
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
import contact.ui.default_config as config
from contact.ui.default_config import format_json_single_line_arrays, loaded_config
from contact.ui.nav_utils import move_highlight, draw_arrows
from contact.utilities.input_handlers import get_list_input
from contact.utilities.singleton import menu_state
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
width = 80
max_help_lines = 6
save_option = "Save Changes"
# Compute an effective width that fits the current terminal
def get_effective_width() -> int:
# Leave space for borders; ensure a sane minimum
return max(20, min(MAX_MENU_WIDTH, curses.COLS - 2))
def edit_color_pair(key: str, current_value: List[str]) -> List[str]:
"""
Allows the user to select a foreground and background color for a key.
@@ -34,14 +27,13 @@ def edit_color_pair(key: str, current_value: List[str]) -> List[str]:
def edit_value(key: str, current_value: str) -> str:
w = get_effective_width()
height = 10
input_width = w - 16 # Allow space for "New Value: "
input_width = width - 16 # Allow space for "New Value: "
start_y = (curses.LINES - height) // 2
start_x = max(0, (curses.COLS - w) // 2)
start_x = (curses.COLS - width) // 2
# Create a centered window
edit_win = curses.newwin(height, w, start_y, start_x)
edit_win = curses.newwin(height, width, start_y, start_x)
edit_win.bkgd(get_color("background"))
edit_win.attrset(get_color("window_frame"))
edit_win.border()
@@ -50,7 +42,7 @@ def edit_value(key: str, current_value: str) -> str:
edit_win.addstr(1, 2, f"Editing {key}", get_color("settings_default", bold=True))
edit_win.addstr(3, 2, "Current Value:", get_color("settings_default"))
wrap_width = w - 4 # Account for border and padding
wrap_width = width - 4 # Account for border and padding
wrapped_lines = [current_value[i : i + wrap_width] for i in range(0, len(current_value), wrap_width)]
for i, line in enumerate(wrapped_lines[:4]): # Limit display to fit the window height
@@ -61,9 +53,7 @@ def edit_value(key: str, current_value: str) -> str:
# Handle theme selection dynamically
if key == "theme":
# Load theme names dynamically from the JSON
theme_options = [
k.split("_", 2)[2].lower() for k in config.loaded_config.keys() if k.startswith("COLOR_CONFIG")
]
theme_options = [k.split("_", 2)[2].lower() for k in loaded_config.keys() if k.startswith("COLOR_CONFIG")]
return get_list_input("Select Theme", current_value, theme_options)
elif key == "node_sort":
@@ -74,10 +64,6 @@ def edit_value(key: str, current_value: str) -> str:
sound_options = ["True", "False"]
return get_list_input("Notification Sound", current_value, sound_options)
elif key == "single_pane_mode":
sound_options = ["True", "False"]
return get_list_input("Single-Pane Mode", current_value, sound_options)
# Standard Input Mode (Scrollable)
edit_win.addstr(7, 2, "New Value: ", get_color("settings_default"))
curses.curs_set(1)
@@ -86,64 +72,41 @@ def edit_value(key: str, current_value: str) -> str:
user_input = ""
input_position = (7, 13) # Tuple for row and column
row, col = input_position # Unpack tuple
while True:
if menu_state.need_redraw:
curses.update_lines_cols()
menu_state.need_redraw = False
# Re-create the window to fully reset state
edit_win = curses.newwin(height, w, start_y, start_x)
edit_win.timeout(200)
edit_win.bkgd(get_color("background"))
edit_win.attrset(get_color("window_frame"))
edit_win.border()
# Redraw static content
edit_win.addstr(1, 2, f"Editing {key}", get_color("settings_default", bold=True))
edit_win.addstr(3, 2, "Current Value:", get_color("settings_default"))
for i, line in enumerate(wrapped_lines[:4]):
edit_win.addstr(4 + i, 2, line, get_color("settings_default"))
edit_win.addstr(7, 2, "New Value: ", get_color("settings_default"))
visible_text = user_input[scroll_offset : scroll_offset + input_width]
edit_win.addstr(row, col, " " * input_width, get_color("settings_default"))
edit_win.addstr(row, col, visible_text, get_color("settings_default"))
visible_text = user_input[scroll_offset : scroll_offset + input_width] # Only show what fits
edit_win.addstr(row, col, " " * input_width, get_color("settings_default")) # Clear previous text
edit_win.addstr(row, col, visible_text, get_color("settings_default")) # Display text
edit_win.refresh()
edit_win.move(row, col + min(len(user_input) - scroll_offset, input_width))
edit_win.move(row, col + min(len(user_input) - scroll_offset, input_width)) # Adjust cursor position
key = edit_win.get_wch()
try:
key = edit_win.get_wch()
except curses.error:
continue # window not ready — skip this loop
if key in (chr(27), curses.KEY_LEFT):
if key in (chr(27), curses.KEY_LEFT): # ESC or Left Arrow
curses.curs_set(0)
return current_value
return current_value # Exit without returning a value
elif key in (chr(curses.KEY_ENTER), chr(10), chr(13)):
break
elif key in (curses.KEY_BACKSPACE, chr(127)):
if user_input:
elif key in (curses.KEY_BACKSPACE, chr(127)): # Backspace
if user_input: # Only process if there's something to delete
user_input = user_input[:-1]
if scroll_offset > 0 and len(user_input) < scroll_offset + input_width:
scroll_offset -= 1
scroll_offset -= 1 # Move back if text is shorter than scrolled area
else:
if isinstance(key, str):
user_input += key
else:
user_input += chr(key)
if len(user_input) > input_width:
if len(user_input) > input_width: # Scroll if input exceeds visible area
scroll_offset += 1
curses.curs_set(0)
return user_input if user_input else current_value
def display_menu() -> tuple[Any, Any, List[str]]:
def display_menu(menu_state: Any) -> tuple[Any, Any, List[str]]:
"""
Render the configuration menu with a Save button directly added to the window.
"""
@@ -161,12 +124,11 @@ def display_menu() -> tuple[Any, Any, List[str]]:
max_menu_height = curses.LINES
menu_height = min(max_menu_height, num_items + 5)
num_items = len(options)
w = get_effective_width()
start_y = (curses.LINES - menu_height) // 2
start_x = max(0, (curses.COLS - w) // 2)
start_x = (curses.COLS - width) // 2
# Create the window
menu_win = curses.newwin(menu_height, w, start_y, start_x)
menu_win = curses.newwin(menu_height, width, start_y, start_x)
menu_win.erase()
menu_win.bkgd(get_color("background"))
menu_win.attrset(get_color("window_frame"))
@@ -174,13 +136,13 @@ def display_menu() -> tuple[Any, Any, List[str]]:
menu_win.keypad(True)
# Create the pad for scrolling
menu_pad = curses.newpad(num_items + 1, w - 8)
menu_pad = curses.newpad(num_items + 1, width - 8)
menu_pad.bkgd(get_color("background"))
# Display the menu path
header = " > ".join(menu_state.menu_path)
if len(header) > w - 4:
header = header[: w - 7] + "..."
if len(header) > width - 4:
header = header[: width - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
# Populate the pad with menu options
@@ -190,18 +152,18 @@ def display_menu() -> tuple[Any, Any, List[str]]:
if isinstance(menu_state.current_menu, dict)
else menu_state.current_menu[int(key.strip("[]"))]
)
display_key = f"{key}"[: w // 2 - 2]
display_value = f"{value}"[: w // 2 - 8]
display_key = f"{key}"[: width // 2 - 2]
display_value = f"{value}"[: width // 2 - 8]
color = get_color("settings_default", reverse=(idx == menu_state.selected_index))
menu_pad.addstr(idx, 0, f"{display_key:<{w // 2 - 2}} {display_value}".ljust(w - 8), color)
menu_pad.addstr(idx, 0, f"{display_key:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
# Add Save button to the main window
if menu_state.show_save_option:
save_position = menu_height - 2
menu_win.addstr(
save_position,
(w - len(save_option)) // 2,
(width - len(save_option)) // 2,
save_option,
get_color("settings_save", reverse=(menu_state.selected_index == len(menu_state.current_menu))),
)
@@ -227,7 +189,6 @@ def display_menu() -> tuple[Any, Any, List[str]]:
def json_editor(stdscr: curses.window, menu_state: Any) -> None:
menu_state.selected_index = 0 # Track the selected option
made_changes = False # Track if any changes were made
script_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.abspath(os.path.join(script_dir, os.pardir))
@@ -250,18 +211,16 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
menu_state.current_menu = data # Track the current level of the menu
# Render the menu
menu_win, menu_pad, options = display_menu()
menu_state.need_redraw = True
menu_win, menu_pad, options = display_menu(menu_state)
need_redraw = True
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
menu_win, menu_pad, options = display_menu()
if need_redraw:
menu_win, menu_pad, options = display_menu(menu_state)
menu_win.refresh()
need_redraw = False
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
menu_win.timeout(200)
key = menu_win.getch()
if key == curses.KEY_UP:
@@ -289,7 +248,7 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
elif key in (curses.KEY_RIGHT, 10, 13): # 10 = \n, 13 = carriage return
menu_state.need_redraw = True
need_redraw = True
menu_win.erase()
menu_win.refresh()
@@ -310,14 +269,11 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
if isinstance(selected_data, list) and len(selected_data) == 2:
# Edit color pair
old = selected_data
new_value = edit_color_pair(selected_key, selected_data)
menu_state.menu_path.pop()
menu_state.start_index.pop()
menu_state.menu_index.pop()
menu_state.current_menu[selected_key] = new_value
if new_value != old:
made_changes = True
elif isinstance(selected_data, (dict, list)):
# Navigate into nested data
@@ -326,27 +282,22 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
else:
# General value editing
old = selected_data
new_value = edit_value(selected_key, selected_data)
menu_state.menu_path.pop()
menu_state.menu_index.pop()
menu_state.start_index.pop()
menu_state.current_menu[selected_key] = new_value
menu_state.need_redraw = True
if new_value != old:
made_changes = True
need_redraw = True
else:
# Save button selected
save_json(file_path, data)
made_changes = False
stdscr.refresh()
# config.reload() # This isn't refreshing the file paths as expected
break
continue
elif key in (27, curses.KEY_LEFT): # Escape or Left Arrow
menu_state.need_redraw = True
need_redraw = True
menu_win.erase()
menu_win.refresh()
@@ -367,19 +318,6 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
else:
# Exit the editor
if made_changes:
save_prompt = get_list_input(
"You have unsaved changes. Save before exiting?",
None,
["Yes", "No", "Cancel"],
mandatory=True,
)
if save_prompt == "Cancel":
continue # Stay in the menu without doing anything
elif save_prompt == "Yes":
save_json(file_path, data)
made_changes = False
menu_win.clear()
menu_win.refresh()
@@ -387,7 +325,7 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
def save_json(file_path: str, data: Dict[str, Any]) -> None:
formatted_json = config.format_json_single_line_arrays(data)
formatted_json = format_json_single_line_arrays(data)
with open(file_path, "w", encoding="utf-8") as f:
f.write(formatted_json)
setup_colors(reinit=True)
@@ -396,6 +334,7 @@ def save_json(file_path: str, data: Dict[str, Any]) -> None:
def main(stdscr: curses.window) -> None:
from contact.ui.ui_state import MenuState
menu_state = MenuState()
if len(menu_state.menu_path) == 0:
menu_state.menu_path = ["App Settings"] # Initialize if not set

View File

@@ -136,18 +136,13 @@ 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"{ts_str} {config.sent_message_prefix}{ack_str}: ",
sanitized_message,
)
formatted_message = (f"{config.sent_message_prefix}{ack_str}: ", sanitized_message)
else:
sanitized_message = message.replace("\x00", "")
formatted_message = (
f"{ts_str} {config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
f"{config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
sanitized_message,
)

View File

@@ -8,20 +8,6 @@ from contact.ui.colors import get_color
from contact.ui.nav_utils import move_highlight, draw_arrows, wrap_text
from contact.ui.dialog import dialog
from contact.utilities.validation_rules import get_validation_for
from contact.utilities.singleton import menu_state
# Dialogs should be at most 80 cols, but shrink on small terminals
MAX_DIALOG_WIDTH = 80
MIN_DIALOG_WIDTH = 20
def get_dialog_width() -> int:
# Leave 2 columns for borders and clamp to a sane minimum
try:
return max(MIN_DIALOG_WIDTH, min(MAX_DIALOG_WIDTH, curses.COLS - 2))
except Exception:
# Fallback if curses not ready yet
return MAX_DIALOG_WIDTH
def invalid_input(window: curses.window, message: str, redraw_func: Optional[callable] = None) -> None:
@@ -58,16 +44,15 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
input_win.refresh()
height = 8
width = get_dialog_width()
width = 80
margin = 2 # Left and right margin
input_width = width - (2 * margin) # Space available for text
max_input_rows = height - 4 # Space for input
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
input_win = curses.newwin(height, width, start_y, start_x)
input_win.timeout(200)
input_win.bkgd(get_color("background"))
input_win.attrset(get_color("window_frame"))
input_win.border()
@@ -105,25 +90,15 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
first_line_width = input_width - len(prompt_text)
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw_input_win()
try:
key = input_win.get_wch()
except curses.error:
continue
key = input_win.get_wch()
if key == chr(27) or key == curses.KEY_LEFT:
input_win.erase()
input_win.refresh()
curses.curs_set(0)
menu_state.need_redraw = True
return None
elif key in (chr(curses.KEY_ENTER), chr(10), chr(13)):
menu_state.need_redraw = True
if not user_input.strip():
invalid_input(input_win, "Value cannot be empty.", redraw_func=redraw_input_win)
continue
@@ -217,7 +192,9 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
# Clear only the input area (without touching prompt text)
for i in range(max_input_rows):
if row + 1 + i < height - 1:
input_win.addstr(row + 1 + i, margin, " " * input_width, get_color("settings_default"))
input_win.addstr(
row + 1 + i, margin, " " * min(input_width, width - margin - 1), get_color("settings_default")
)
# Redraw the prompt text so it never disappears
input_win.addstr(row + 1, margin, prompt_text, get_color("settings_default"))
@@ -255,15 +232,14 @@ def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
cvalue = to_base64(current_value) # Convert current values to Base64
height = 9
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
admin_key_win = curses.newwin(height, width, start_y, start_x)
admin_key_win.timeout(200)
admin_key_win.bkgd(get_color("background"))
admin_key_win.attrset(get_color("window_frame"))
admin_key_win.keypad(True) # Enable keypad for special keys
repeated_win = curses.newwin(height, width, start_y, start_x)
repeated_win.bkgd(get_color("background"))
repeated_win.attrset(get_color("window_frame"))
repeated_win.keypad(True) # Enable keypad for special keys
curses.echo()
curses.curs_set(1)
@@ -274,39 +250,37 @@ def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
invalid_input = ""
while True:
admin_key_win.erase()
admin_key_win.border()
admin_key_win.addstr(1, 2, "Edit up to 3 Admin Keys:", get_color("settings_default", bold=True))
repeated_win.erase()
repeated_win.border()
repeated_win.addstr(1, 2, "Edit up to 3 Admin Keys:", get_color("settings_default", bold=True))
# Display current values, allowing editing
for i, line in enumerate(user_values):
prefix = "" if i == cursor_pos else " " # Highlight the current line
admin_key_win.addstr(
repeated_win.addstr(
3 + i, 2, f"{prefix}Admin Key {i + 1}: ", get_color("settings_default", bold=(i == cursor_pos))
)
admin_key_win.addstr(3 + i, 18, line) # Align text for easier editing
repeated_win.addstr(3 + i, 18, line) # Align text for easier editing
# Move cursor to the correct position inside the field
curses.curs_set(1)
admin_key_win.move(3 + cursor_pos, 18 + len(user_values[cursor_pos])) # Position cursor at end of text
repeated_win.move(3 + cursor_pos, 18 + len(user_values[cursor_pos])) # Position cursor at end of text
# Show error message if needed
if invalid_input:
admin_key_win.addstr(7, 2, invalid_input, get_color("settings_default", bold=True))
repeated_win.addstr(7, 2, invalid_input, get_color("settings_default", bold=True))
admin_key_win.refresh()
key = admin_key_win.getch()
repeated_win.refresh()
key = repeated_win.getch()
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow -> Cancel and return original
admin_key_win.erase()
admin_key_win.refresh()
repeated_win.erase()
repeated_win.refresh()
curses.noecho()
curses.curs_set(0)
menu_state.need_redraw = True
return None
elif key == ord("\n"): # Enter key to save and return
menu_state.need_redraw = True
if all(is_valid_base64(val) for val in user_values): # Ensure all values are valid Base64 and 32 bytes
curses.noecho()
curses.curs_set(0)
@@ -328,247 +302,195 @@ def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
pass # Ignore invalid character inputs
from contact.utilities.singleton import menu_state # Required if not already imported
def get_repeated_input(current_value: List[str]) -> Optional[str]:
height = 9
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
repeated_win = curses.newwin(height, width, start_y, start_x)
repeated_win.timeout(200)
repeated_win.bkgd(get_color("background"))
repeated_win.attrset(get_color("window_frame"))
repeated_win.keypad(True)
repeated_win.keypad(True) # Enable keypad for special keys
curses.echo()
curses.curs_set(1)
curses.curs_set(1) # Show the cursor
user_values = current_value[:3] + [""] * (3 - len(current_value)) # Always 3 fields
cursor_pos = 0
# Editable list of values (max 3 values)
user_values = current_value[:3]
cursor_pos = 0 # Track which value is being edited
invalid_input = ""
def redraw():
while True:
repeated_win.erase()
repeated_win.border()
repeated_win.addstr(1, 2, "Edit up to 3 Values:", get_color("settings_default", bold=True))
win_h, win_w = repeated_win.getmaxyx()
# Display current values, allowing editing
for i, line in enumerate(user_values):
prefix = "" if i == cursor_pos else " "
prefix = "" if i == cursor_pos else " " # Highlight the current line
repeated_win.addstr(
3 + i, 2, f"{prefix}Value{i + 1}: ", get_color("settings_default", bold=(i == cursor_pos))
)
repeated_win.addstr(3 + i, 18, line[: max(0, win_w - 20)]) # Prevent overflow
repeated_win.addstr(3 + i, 18, line)
# Move cursor to the correct position inside the field
curses.curs_set(1)
repeated_win.move(3 + cursor_pos, 18 + len(user_values[cursor_pos])) # Position cursor at end of text
# Show error message if needed
if invalid_input:
win_h, win_w = repeated_win.getmaxyx()
repeated_win.addstr(7, 2, invalid_input[: max(0, win_w - 4)], get_color("settings_default", bold=True))
repeated_win.addstr(7, 2, invalid_input, get_color("settings_default", bold=True))
repeated_win.move(3 + cursor_pos, 18 + len(user_values[cursor_pos]))
repeated_win.refresh()
key = repeated_win.getch()
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw()
redraw()
try:
key = repeated_win.get_wch()
except curses.error:
continue # ignore timeout or input issues
if key in (27, curses.KEY_LEFT): # ESC or Left Arrow
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow -> Cancel and return original
repeated_win.erase()
repeated_win.refresh()
curses.noecho()
curses.curs_set(0)
menu_state.need_redraw = True
return None
elif key in ("\n", curses.KEY_ENTER):
elif key == ord("\n"): # Enter key to save and return
curses.noecho()
curses.curs_set(0)
menu_state.need_redraw = True
return ", ".join(user_values).strip()
elif key == curses.KEY_UP:
cursor_pos = (cursor_pos - 1) % 3
elif key == curses.KEY_DOWN:
cursor_pos = (cursor_pos + 1) % 3
elif key in (curses.KEY_BACKSPACE, 127):
user_values[cursor_pos] = user_values[cursor_pos][:-1]
return ", ".join(user_values)
elif key == curses.KEY_UP: # Move cursor up
cursor_pos = (cursor_pos - 1) % len(user_values)
elif key == curses.KEY_DOWN: # Move cursor down
cursor_pos = (cursor_pos + 1) % len(user_values)
elif key == curses.KEY_BACKSPACE or key == 127: # Backspace key
if len(user_values[cursor_pos]) > 0:
user_values[cursor_pos] = user_values[cursor_pos][:-1] # Remove last character
else:
try:
ch = chr(key) if isinstance(key, int) else key
if ch.isprintable():
user_values[cursor_pos] += ch
invalid_input = ""
except Exception:
pass
from contact.utilities.singleton import menu_state # Ensure this is imported
user_values[cursor_pos] += chr(key) # Append valid character input to the selected field
invalid_input = "" # Clear error if user starts fixing input
except ValueError:
pass # Ignore invalid character inputs
def get_fixed32_input(current_value: int) -> int:
original_value = current_value
ip_string = str(ipaddress.IPv4Address(current_value))
cvalue = current_value
current_value = str(ipaddress.IPv4Address(current_value))
height = 10
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
fixed32_win = curses.newwin(height, width, start_y, start_x)
fixed32_win.bkgd(get_color("background"))
fixed32_win.attrset(get_color("window_frame"))
fixed32_win.keypad(True)
fixed32_win.timeout(200)
curses.echo()
curses.curs_set(1)
user_input = ""
def redraw():
while True:
fixed32_win.erase()
fixed32_win.border()
fixed32_win.addstr(1, 2, "Enter an IP address (xxx.xxx.xxx.xxx):", get_color("settings_default", bold=True))
fixed32_win.addstr(3, 2, f"Current: {ip_string}", get_color("settings_default"))
fixed32_win.addstr(5, 2, f"New value: {user_input}", get_color("settings_default"))
fixed32_win.addstr(1, 2, "Enter an IP address (xxx.xxx.xxx.xxx):", curses.A_BOLD)
fixed32_win.addstr(3, 2, f"Current: {current_value}")
fixed32_win.addstr(5, 2, f"New value: {user_input}")
fixed32_win.refresh()
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw()
key = fixed32_win.getch()
redraw()
try:
key = fixed32_win.get_wch()
except curses.error:
continue # ignore timeout
if key in (27, curses.KEY_LEFT): # ESC or Left Arrow to cancel
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow to cancel
fixed32_win.erase()
fixed32_win.refresh()
curses.noecho()
curses.curs_set(0)
menu_state.need_redraw = True
return original_value
elif key in ("\n", curses.KEY_ENTER):
return cvalue # Return the current value unchanged
elif key == ord("\n"): # Enter key to validate and save
# Validate IP address
octets = user_input.split(".")
menu_state.need_redraw = True
if len(octets) == 4 and all(octet.isdigit() and 0 <= int(octet) <= 255 for octet in octets):
curses.noecho()
curses.curs_set(0)
return int(ipaddress.ip_address(user_input))
fixed32_address = ipaddress.ip_address(user_input)
return int(fixed32_address) # Return the valid IP address
else:
fixed32_win.addstr(7, 2, "Invalid IP address. Try again.", get_color("settings_default", bold=True))
fixed32_win.addstr(7, 2, "Invalid IP address. Try again.", curses.A_BOLD | curses.color_pair(5))
fixed32_win.refresh()
curses.napms(1500)
user_input = ""
elif key in (curses.KEY_BACKSPACE, 127):
curses.napms(1500) # Wait for 1.5 seconds before refreshing
user_input = "" # Clear invalid input
elif key == curses.KEY_BACKSPACE or key == 127: # Backspace key
user_input = user_input[:-1]
else:
try:
ch = chr(key) if isinstance(key, int) else key
if ch.isdigit() or ch == ".":
user_input += ch
except Exception:
pass # Ignore unprintable inputs
char = chr(key)
if char.isdigit() or char == ".":
user_input += char # Append only valid characters (digits or dots)
except ValueError:
pass # Ignore invalid inputs
from typing import List, Optional # ensure Optional is imported
def get_list_input(
prompt: str, current_option: Optional[str], list_options: List[str], mandatory: bool = False
) -> Optional[str]:
def get_list_input(prompt: str, current_option: Optional[str], list_options: List[str]) -> Optional[str]:
"""
List selector.
Displays a scrollable list of list_options for the user to choose from.
"""
selected_index = list_options.index(current_option) if current_option in list_options else 0
height = min(len(list_options) + 5, curses.LINES)
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
list_win = curses.newwin(height, width, start_y, start_x)
list_win.timeout(200)
list_win.bkgd(get_color("background"))
list_win.attrset(get_color("window_frame"))
list_win.keypad(True)
list_pad = curses.newpad(len(list_options) + 1, max(1, width - 8))
list_pad = curses.newpad(len(list_options) + 1, width - 8)
list_pad.bkgd(get_color("background"))
# Render header
list_win.erase()
list_win.border()
list_win.addstr(1, 2, prompt, get_color("settings_default", bold=True))
# Render options on the pad
for idx, color in enumerate(list_options):
if idx == selected_index:
list_pad.addstr(idx, 0, color.ljust(width - 8), get_color("settings_default", reverse=True))
else:
list_pad.addstr(idx, 0, color.ljust(width - 8), get_color("settings_default"))
# Initial refresh
list_win.refresh()
list_pad.refresh(
0,
0,
list_win.getbegyx()[0] + 3,
list_win.getbegyx()[1] + 4,
list_win.getbegyx()[0] + list_win.getmaxyx()[0] - 2,
list_win.getbegyx()[1] + list_win.getmaxyx()[1] - 4,
)
max_index = len(list_options) - 1
visible_height = list_win.getmaxyx()[0] - 5
def redraw_list_ui():
list_win.erase()
list_win.border()
list_win.addstr(1, 2, prompt, get_color("settings_default", bold=True))
win_h, win_w = list_win.getmaxyx()
pad_w = max(1, win_w - 8)
for idx, item in enumerate(list_options):
color = get_color("settings_default", reverse=(idx == selected_index))
list_pad.addstr(idx, 0, item[:pad_w].ljust(pad_w), color)
list_win.refresh()
list_pad.refresh(
0,
0,
list_win.getbegyx()[0] + 3,
list_win.getbegyx()[1] + 4,
list_win.getbegyx()[0] + list_win.getmaxyx()[0] - 2,
list_win.getbegyx()[1] + list_win.getmaxyx()[1] - 4,
)
# Recompute visible height each draw in case of resize
vis_h = list_win.getmaxyx()[0] - 5
draw_arrows(list_win, vis_h, max_index, [0], show_save_option=False)
# Initial draw
redraw_list_ui()
draw_arrows(list_win, visible_height, max_index, [0], show_save_option=False) # Initial call to draw arrows
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw_list_ui()
try:
key = list_win.getch()
except curses.error:
continue
key = list_win.getch()
if key == curses.KEY_UP:
old_selected_index = selected_index
selected_index = max(0, selected_index - 1)
move_highlight(old_selected_index, list_options, list_win, list_pad, selected_index=selected_index)
elif key == curses.KEY_DOWN:
old_selected_index = selected_index
selected_index = min(len(list_options) - 1, selected_index + 1)
move_highlight(old_selected_index, list_options, list_win, list_pad, selected_index=selected_index)
elif key == ord("\n"): # Enter
elif key == ord("\n"): # Enter key
list_win.clear()
list_win.refresh()
menu_state.need_redraw = True
return list_options[selected_index]
elif key == 27 or key == curses.KEY_LEFT: # ESC or Left
if mandatory:
continue
elif key == 27 or key == curses.KEY_LEFT: # ESC or Left Arrow
list_win.clear()
list_win.refresh()
menu_state.need_redraw = True
return current_option

View File

@@ -1,6 +1,5 @@
from contact.ui.ui_state import ChatUIState, InterfaceState, AppState, MenuState
from contact.ui.ui_state import ChatUIState, InterfaceState, AppState
ui_state = ChatUIState()
interface_state = InterfaceState()
app_state = AppState()
menu_state = MenuState()

View File

@@ -1,90 +0,0 @@
import datetime
sensors = {
'temperature': {'icon':'🌡️ ','unit':'°'},
'relative_humidity': {'icon':'💧','unit':'%'},
'barometric_pressure': {'icon':'','unit': 'hPa'},
'lux': {'icon':'🔦 ','unit': 'lx'},
'uv_lux': {'icon':'uv🔦 ','unit': 'lx'},
'wind_speed': {'icon':'💨 ','unit': 'm/s'},
'wind_direction': {'icon':'','unit': ''},
'battery_level': {'icon':'🔋 ', 'unit':'%'},
'voltage': {'icon':'', 'unit':'V'},
'channel_utilization': {'icon':'ChUtil:', 'unit':'%'},
'air_util_tx': {'icon':'AirUtil:', 'unit':'%'},
'uptime_seconds': {'icon':'🆙 ', 'unit':'h'},
'latitude_i': {'icon':'🌍 ', 'unit':''},
'longitude_i': {'icon':'', 'unit':''},
'altitude': {'icon':'⬆️ ', 'unit':'m'},
'time': {'icon':'🕔 ', 'unit':''}
}
def humanize_wind_direction(degrees):
""" Convert degrees to Eest-West-Nnoth-Ssouth directions """
if not 0 <= degrees <= 360:
return None
directions = [
("N", 337.5, 22.5),
("NE", 22.5, 67.5),
("E", 67.5, 112.5),
("SE", 112.5, 157.5),
("S", 157.5, 202.5),
("SW", 202.5, 247.5),
("W", 247.5, 292.5),
("NW", 292.5, 337.5),
]
if degrees >= directions[0][1] or degrees < directions[0][2]:
return directions[0][0]
# Check for all other directions
for direction, lower_bound, upper_bound in directions[1:]:
if lower_bound <= degrees < upper_bound:
return direction
# This part should ideally not be reached with valid input
return None
def get_chunks(data):
""" Breakdown telemetry data and assign emojis for more visual appeal of the payloads """
reading = data.split('\n')
# remove empty list lefover from the split
reading = list(filter(None, reading))
parsed=""
for item in reading:
key, value = item.split(":")
# If value is float, round it to the 1 digit after point
# else make it int
if "." in value:
value = round(float(value.strip()),1)
else:
try:
value = int(value.strip())
except Exception:
# Leave it string as last resort
value = value
match key:
# convert seconds to hours, for our sanity
case "uptime_seconds":
value = round(value / 60 / 60, 1)
# Convert position to degrees (humanize), as per Meshtastic protobuf comment for this telemetry
# truncate to 6th digit after floating point, which would be still accurate
case "longitude_i" | "latitude_i":
value = round(value * 1e-7, 6)
# Convert wind direction from degrees to abbreviation
case "wind_direction":
value = humanize_wind_direction(value)
case "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
if key in sensors:
parsed+= f"{sensors[key.strip()]['icon']}{value}{sensors[key]['unit']} "
else:
# just pass through if we haven't added the particular telemetry key:value to the sensor dict
parsed+=f"{key}:{value} "
return parsed

View File

@@ -1,13 +1,9 @@
import datetime
import time
from typing import Optional, Union
from google.protobuf.message import DecodeError
from meshtastic import protocols
from meshtastic.protobuf import config_pb2, mesh_pb2, portnums_pb2
from meshtastic.protobuf import config_pb2
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
import contact.utilities.telemetry_beautifier as tb
def get_channels():
@@ -140,7 +136,6 @@ def get_time_ago(timestamp):
return f"{value} {unit} ago"
return "now"
def add_new_message(channel_id, prefix, message):
if channel_id not in ui_state.all_messages:
ui_state.all_messages[channel_id] = []
@@ -167,53 +162,4 @@ def add_new_message(channel_id, prefix, message):
ui_state.all_messages[channel_id].append((f"-- {current_hour} --", ""))
# Add the 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]:
"""Attempt to parse a decoded payload using the registered protobuf handler."""
try:
decoded = packet.get("decoded") or {}
portnum = decoded.get("portnum")
payload = decoded.get("payload")
if isinstance(payload, str):
return payload
# These portnumbers carry information visible elswhere in the app, so we just note them in the logs
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:
try:
pb = handler.protobufFactory()
pb.ParseFromString(bytes(payload))
# If we have position payload
if portnum == "POSITION_APP":
return tb.get_chunks(str(pb))
# Part of TELEMETRY_APP portnum
if hasattr(pb, "device_metrics") and pb.HasField("device_metrics"):
return tb.get_chunks(str(pb.device_metrics))
# Part of TELEMETRY_APP portnum
if hasattr(pb, "environment_metrics") and pb.HasField("environment_metrics"):
return tb.get_chunks(str(pb.environment_metrics))
# For other data, without implemented beautification, fallback to just printing the object
return str(pb).replace("\n", " ").replace("\r", " ").strip()
except DecodeError:
return payload
# return payload
except Exception:
return payload
ui_state.all_messages[channel_id].append((prefix,message))

View File

@@ -1,13 +1,13 @@
[project]
name = "contact"
version = "1.4.8"
version = "1.3.16"
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"}
]
license = "GPL-3.0-only"
readme = "README.md"
requires-python = ">=3.9,<4.0"
requires-python = ">=3.9,<3.14"
dependencies = [
"meshtastic (>=2.6.0,<3.0.0)"
]