Compare commits

...

64 Commits

Author SHA1 Message Date
pdxlocations
68578b534a hide help on small screens 2025-08-22 14:39:43 -07:00
pdxlocations
27b0ca7482 allow smaller windows 2025-08-22 12:17:34 -07:00
pdxlocations
d939561e2d refactor 2025-08-22 11:29:30 -07:00
pdxlocations
9ab3c4dbf9 Bonus, redraw settings when new line in packetlog 2025-08-21 23:15:45 -07:00
pdxlocations
d79b751599 fix packet log crash 2025-08-21 23:05:27 -07:00
pdxlocations
0a00e08431 fix single-pane crash 2025-08-21 22:45:56 -07:00
pdxlocations
f2627e0ed4 focus arrows fix 2025-08-21 00:18:53 -07:00
pdxlocations
a3f0232641 fix save check 2025-08-21 00:05:06 -07:00
pdxlocations
3987c183d8 shift focus on message send 2025-08-20 23:54:36 -07:00
pdxlocations
fcb1a42ef2 init 2025-08-20 23:49:17 -07:00
pdxlocations
751a143d0a Merge pull request #216 from jekeam/patch-1 2025-08-13 07:25:44 -07:00
jekeam
f7d203e97a Update README.md [Install for Window] 2025-08-13 14:48:05 +05:00
pdxlocations
de4f813b90 bump version 2025-08-12 21:55:34 -07:00
pdxlocations
e17f7e576f hide cursor 2025-08-12 21:48:21 -07:00
pdxlocations
dccdb00dcd Redraw settings menu on new node 2025-08-12 21:32:38 -07:00
pdxlocations
81fd7a26f5 Merge pull request #215 from pdxlocations:confirm-unsaved-changes
Confirm unsaved Changes
2025-08-08 00:39:17 -07:00
pdxlocations
640955656f fix no config redraw 2025-08-08 00:38:34 -07:00
pdxlocations
8f248f4b5b add confirmation to app settings 2025-08-08 00:29:11 -07:00
pdxlocations
c10905e954 don't exit dialog with left arrow 2025-08-07 23:58:47 -07:00
pdxlocations
d1b93263fa add confirmation box if settings not saved 2025-08-07 23:34:36 -07:00
pdxlocations
623708c2a1 update README.md 2025-08-07 22:30:18 -07:00
pdxlocations
9b8abdb344 fix admin key window name 2025-07-31 23:48:15 -07:00
pdxlocations
8c3e00b52b redraw other input types 2025-07-31 23:44:14 -07:00
pdxlocations
81ebd1b95f fix get_text_input 2025-07-31 00:55:19 -07:00
pdxlocations
ae75d85741 fix user settings inputs 2025-07-31 00:33:30 -07:00
pdxlocations
b6767f423e Fix settings redraw (#214)
* current window 4

* refresh settings on new message

* redraw dialog and fix traceroute

* formatting and catch

* move continue
2025-07-31 00:08:08 -07:00
pdxlocations
b1252fec6c Update README.md 2025-07-30 22:18:01 -07:00
pdxlocations
43d1152074 Update README.md 2025-07-30 22:14:46 -07:00
pdxlocations
786a7b03c5 Configure Filepath for Export Node Config (#213)
* add node config path to settings

* try reload config but failed
2025-07-29 16:51:26 -07:00
pdxlocations
8d111c5df7 Add Warning for Sending Messages Quickly (#212)
* Warn About 2-Second Message Delay

* add comment

* update lines and cols
2025-07-28 23:04:57 -07:00
pdxlocations
b314a24a0c Input Validation Framework (#211)
* init

* validation framework

* add rules

* automatic types

* changes

* fix positions

* redraw input

* check for selected_config

* tweaks

* refactor
2025-07-26 21:20:15 -07:00
pdxlocations
4378f3045c unused argument 2025-07-24 18:02:44 -07:00
pdxlocations
a451d1d7d6 note to future me 2025-07-21 23:18:11 -07:00
pdxlocations
fe1f027219 Merge pull request #209 from pdxlocations/check-db-fields-for-null
Check for NULLS in DB
2025-07-21 10:57:18 -07:00
pdxlocations
43435cbe04 replace \x00 in messages 2025-07-21 10:54:19 -07:00
pdxlocations
fe98075582 bump version 2025-07-21 00:04:00 -07:00
pdxlocations
8716ea6fe1 dont write to the log before config 2025-07-20 23:46:24 -07:00
pdxlocations
a8bdcbb7e6 Merge pull request #208 from pdxlocations/config
fallback to user if install dir not writable
2025-07-17 23:00:13 -07:00
pdxlocations
02742b27f3 fallback to user if install dir not writable 2025-07-17 22:59:35 -07:00
pdxlocations
ae028032a0 Merge pull request #207 from pdxlocations/errors
show connection errors in console
2025-07-17 00:12:12 -07:00
pdxlocations
30402f4906 show connection errors in console 2025-07-17 00:10:17 -07:00
pdxlocations
324e0b03e7 Merge pull request #206 from pdxlocations/notifications
maybe fix aplay
2025-07-16 19:43:37 -07:00
pdxlocations
056db12911 maybe fix aplay 2025-07-16 18:46:51 -07:00
pdxlocations
685a2d4bf8 bump version 2025-07-14 08:15:16 -07:00
pdxlocations
6ed0cc8c9f Merge pull request #199 from rfschmid/add-traceroute-sent-message-to-history
Add traceroute sent message to history
2025-07-03 10:46:07 -07:00
Russell Schmidt
fc208a9258 Add "Traceroute Sent" to message history 2025-07-03 12:43:18 -05:00
Russell Schmidt
eaf9381bca Refactor message saving
Add common function for saving a message to history, removing some
duplicate code and making traceroutes add timestamps like other messages
do.
2025-07-03 12:43:18 -05:00
pdxlocations
367af9044c Merge pull request #198 from rfschmid/add-node-name-to-traceroute-confirm-dialog
Add node name to traceroute confirm dialog
2025-07-03 10:32:41 -07:00
Russell Schmidt
d8183d9009 Make capitalization consistent 2025-07-03 12:19:24 -05:00
Russell Schmidt
3fb1335be3 Add node name to traceroute confirm dialog 2025-07-03 12:10:56 -05:00
pdxlocations
8b05072786 bump version 2025-06-13 15:33:57 -07:00
pdxlocations
4455781e6c fix types and returns 2025-06-12 16:38:05 -07:00
pdxlocations
0c8aaee415 Merge pull request #197 from rfschmid/redirect-sound-player-output-to-dev-null
Redirect sound player output to dev null
2025-06-12 16:10:26 -07:00
pdxlocations
b97d9f4649 Merge pull request #196 from rfschmid/only-clear-input-text-on-enter-if-sending-message
Only clear input on enter when sending message
2025-06-12 16:09:44 -07:00
Russell Schmidt
4152fb6a21 Redirect sound player output to dev null
On my linux system, the sound playing code goes to aplay. When called,
aplay outputs a message about the file it is playing to stderr, which
causes it to be printed on the input line, which can't be easily
cleared. Redirect output from the audio player executable to dev/null.
Deduplicate sound playing code a bit so we only need one call to
subprocess.run, so I don't have to make this change in three places.
2025-06-12 17:28:30 -05:00
Russell Schmidt
384e36dac2 Only clear input on enter when sending message
We should only clear the input field when the user presses enter if the
user actually sent the message. If selecting a different node to send
to, don't clear input.
2025-06-12 17:23:03 -05:00
pdxlocations
65bca84fe6 minor refactor 2025-06-10 23:24:11 -07:00
pdxlocations
16fa2830fd bump version 2025-06-10 22:29:31 -07:00
pdxlocations
c8f1da99e3 Merge pull request #194 from rfschmid:fix-crash-with-newlines
Fix crash with newlines, message spacing
2025-06-10 22:28:41 -07:00
Russell Schmidt
702250c329 Fix crash with newlines, message spacing 2025-06-10 17:42:14 -05:00
pdxlocations
6291082405 Merge pull request #192 from rfschmid/fix-wrapping-with-wide-chars
Fix crash when wrapping with wide characters
2025-06-10 12:19:02 -07:00
pdxlocations
4fa5148664 Merge pull request #193 from rfschmid/fix-backspace
Fix enter not clearing input
2025-06-10 11:57:06 -07:00
Russell Schmidt
d62ec09eea Fix enter not clearing input
Similar to 981d72e, pressing enter wasn't clearing the input field.
2025-06-10 12:19:03 -05:00
Russell Schmidt
61026dcc73 Fix crash when wrapping with wide characters
Update contact_ui.py to use already-existing custom wrap function
implemented in nav_utils instead of textwrap library. Update custom
wrap_text function to use east_asian_width to determine characters that
can use two columns of width.
2025-06-10 12:17:23 -05:00
17 changed files with 1112 additions and 521 deletions

View File

@@ -6,6 +6,13 @@
```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.
@@ -33,6 +40,7 @@ By navigating to Settings -> App Settings, you may customize your UI's icons, co
- `CTRL` + `p` = Hide/show a log of raw received packets.
- `CTRL` + `t` = With the Node List highlighted, send a traceroute to the selected node
- `CTRL` + `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
@@ -62,8 +70,17 @@ 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

@@ -53,26 +53,26 @@ logging.basicConfig(
app_state.lock = threading.Lock()
# ------------------------------------------------------------------------------
# Main Program Logic
# ------------------------------------------------------------------------------
def prompt_region_if_unset(args: object) -> None:
"""Prompt user to set region if it is unset."""
confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"])
if confirmation == "Yes":
set_region(interface_state.interface)
interface_state.interface.close()
interface_state.interface = initialize_interface(args)
def initialize_globals(args) -> None:
def initialize_globals() -> None:
"""Initializes interface and shared globals."""
interface_state.interface = initialize_interface(args)
# Prompt for region if unset
if interface_state.interface.localNode.localConfig.lora.region == 0:
confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"])
if confirmation == "Yes":
set_region(interface_state.interface)
interface_state.interface.close()
interface_state.interface = initialize_interface(args)
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()
@@ -81,55 +81,63 @@ def initialize_globals(args) -> None:
def main(stdscr: curses.window) -> None:
"""Main entry point for the curses UI."""
output_capture = io.StringIO()
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
setup_colors()
draw_splash(stdscr)
setup_colors()
draw_splash(stdscr)
args = setup_parser().parse_args()
args = setup_parser().parse_args()
if getattr(args, "settings", False):
subprocess.run([sys.executable, "-m", "contact.settings"], check=True)
return
if getattr(args, "settings", False):
subprocess.run([sys.executable, "-m", "contact.settings"], check=True)
return
logging.info("Initializing interface...")
with app_state.lock:
initialize_globals(args)
logging.info("Starting main UI")
logging.info("Initializing interface...")
with app_state.lock:
interface_state.interface = initialize_interface(args)
main_ui(stdscr)
if interface_state.interface.localNode.localConfig.lora.region == 0:
prompt_region_if_unset(args)
except Exception as e:
console_output = output_capture.getvalue()
logging.error("Uncaught exception: %s", e)
logging.error("Traceback: %s", traceback.format_exc())
logging.error("Console output:\n%s", console_output)
initialize_globals()
logging.info("Starting main UI")
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
main_ui(stdscr)
except Exception:
console_output = output_capture.getvalue()
logging.error("Uncaught exception inside main_ui")
logging.error("Traceback:\n%s", traceback.format_exc())
logging.error("Console output:\n%s", console_output)
return
except Exception:
raise
def start() -> None:
"""Launch curses wrapper and redirect logs to file."""
"""Entry point for the application."""
if "--help" in sys.argv or "-h" in sys.argv:
setup_parser().print_help()
sys.exit(0)
with open(config.log_file_path, "a", buffering=1) as log_f:
sys.stdout = log_f
sys.stderr = log_f
with contextlib.redirect_stdout(log_f), contextlib.redirect_stderr(log_f):
try:
curses.wrapper(main)
except KeyboardInterrupt:
logging.info("User exited with Ctrl+C")
sys.exit(0)
except Exception as e:
logging.error("Fatal error: %s", e)
logging.error("Traceback: %s", traceback.format_exc())
sys.exit(1)
try:
curses.wrapper(main)
except KeyboardInterrupt:
logging.info("User exited with Ctrl+C")
sys.exit(0)
except Exception as e:
logging.critical("Fatal error", exc_info=True)
try:
curses.endwin()
except Exception:
pass
print("Fatal error:", e)
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":

View File

@@ -3,11 +3,12 @@ import os
import platform
import shutil
import subprocess
import time
from datetime import datetime
from typing import Any, Dict
from contact.utilities.utils import refresh_node_list
from contact.utilities.utils import (
refresh_node_list,
add_new_message,
)
from contact.ui.contact_ui import (
draw_packetlog_win,
draw_node_list,
@@ -23,43 +24,48 @@ from contact.utilities.db_handler import (
)
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state, app_state
from contact.utilities.singleton import ui_state, interface_state, app_state, menu_state
def play_sound():
try:
system = platform.system()
sound_path = None
executable = None
if system == "Darwin": # macOS
sound_path = "/System/Library/Sounds/Ping.aiff"
if os.path.exists(sound_path):
subprocess.run(["afplay", sound_path], check=True)
return
else:
logging.warning(f"macOS sound file not found: {sound_path}")
executable = "afplay"
elif system == "Linux":
sound_path = "/usr/share/sounds/freedesktop/stereo/complete.oga"
if os.path.exists(sound_path):
if shutil.which("paplay"):
subprocess.run(["paplay", sound_path], check=True)
return
elif shutil.which("aplay"):
subprocess.run(["aplay", sound_path], check=True)
return
else:
logging.warning("No sound player found (paplay/aplay)")
ogg_path = "/usr/share/sounds/freedesktop/stereo/complete.oga"
wav_path = "/usr/share/sounds/alsa/Front_Center.wav" # common fallback
if shutil.which("paplay") and os.path.exists(ogg_path):
executable = "paplay"
sound_path = ogg_path
elif shutil.which("ffplay") and os.path.exists(ogg_path):
executable = "ffplay"
sound_path = ogg_path
elif shutil.which("aplay") and os.path.exists(wav_path):
executable = "aplay"
sound_path = wav_path
else:
logging.warning(f"Linux sound file not found: {sound_path}")
logging.warning("No suitable sound player or sound file found on Linux")
if executable and sound_path:
cmd = [executable, sound_path]
if executable == "ffplay":
cmd = [executable, "-nodisp", "-autoexit", sound_path]
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return
except subprocess.CalledProcessError as e:
logging.error(f"Sound playback failed: {e}")
except Exception as e:
logging.error(f"Unexpected error: {e}")
# Final fallback: terminal beep
print("\a")
def on_receive(packet: Dict[str, Any], interface: Any) -> None:
"""
@@ -78,6 +84,9 @@ 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
@@ -119,7 +128,9 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
channel_number = ui_state.channel_list.index(packet["from"])
if ui_state.channel_list[channel_number] != ui_state.channel_list[ui_state.selected_channel]:
channel_id = ui_state.channel_list[channel_number]
if channel_id != ui_state.channel_list[ui_state.selected_channel]:
add_notification(channel_number)
refresh_channels = True
else:
@@ -129,40 +140,14 @@ 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") + ":"
if ui_state.channel_list[channel_number] not in ui_state.all_messages:
ui_state.all_messages[ui_state.channel_list[channel_number]] = []
# Timestamp handling
current_timestamp = time.time()
current_hour = datetime.fromtimestamp(current_timestamp).strftime("%Y-%m-%d %H:00")
# Retrieve the last timestamp if available
channel_messages = ui_state.all_messages[ui_state.channel_list[channel_number]]
if channel_messages:
# Check the last entry for a timestamp
for entry in reversed(channel_messages):
if entry[0].startswith("--"):
last_hour = entry[0].strip("- ").strip()
break
else:
last_hour = None
else:
last_hour = None
# Add a new timestamp if it's a new hour
if last_hour != current_hour:
ui_state.all_messages[ui_state.channel_list[channel_number]].append((f"-- {current_hour} --", ""))
ui_state.all_messages[ui_state.channel_list[channel_number]].append(
(f"{config.message_prefix} {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()
if refresh_messages:
draw_messages_window(True)
save_message_to_db(ui_state.channel_list[channel_number], message_from_id, message_string)
save_message_to_db(channel_id, message_from_id, message_string)
except KeyError as e:
logging.error(f"Error processing packet: {e}")

View File

@@ -1,4 +1,3 @@
from datetime import datetime
from typing import Any, Dict
import google.protobuf.json_format
@@ -16,6 +15,8 @@ import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
from contact.utilities.utils import add_new_message
ack_naks: Dict[str, Dict[str, Any]] = {} # requestId -> {channel, messageIndex, timestamp}
@@ -146,8 +147,9 @@ def on_response_traceroute(packet: Dict[str, Any]) -> None:
update_node_info_in_db(packet["from"], chat_archived=False)
channel_number = ui_state.channel_list.index(packet["from"])
channel_id = ui_state.channel_list[channel_number]
if ui_state.channel_list[channel_number] == ui_state.channel_list[ui_state.selected_channel]:
if channel_id == ui_state.channel_list[ui_state.selected_channel]:
refresh_messages = True
else:
add_notification(channel_number)
@@ -155,18 +157,14 @@ def on_response_traceroute(packet: Dict[str, Any]) -> None:
message_from_string = get_name_from_database(packet["from"], type="short") + ":\n"
if ui_state.channel_list[channel_number] not in ui_state.all_messages:
ui_state.all_messages[ui_state.channel_list[channel_number]] = []
ui_state.all_messages[ui_state.channel_list[channel_number]].append(
(f"{config.message_prefix} {message_from_string}", msg_str)
)
add_new_message(channel_id, f"{config.message_prefix} {message_from_string}", msg_str)
if refresh_channels:
draw_channel_list()
if refresh_messages:
draw_messages_window(True)
save_message_to_db(ui_state.channel_list[channel_number], packet["from"], msg_str)
save_message_to_db(channel_id, packet["from"], msg_str)
def send_message(message: str, destination: int = BROADCAST_NUM, channel: int = 0) -> None:
"""
@@ -190,32 +188,7 @@ def send_message(message: str, destination: int = BROADCAST_NUM, channel: int =
channelIndex=send_on_channel,
)
# Add sent message to the messages dictionary
if channel_id not in ui_state.all_messages:
ui_state.all_messages[channel_id] = []
# Handle timestamp logic
current_timestamp = int(datetime.now().timestamp()) # Get current timestamp
current_hour = datetime.fromtimestamp(current_timestamp).strftime("%Y-%m-%d %H:00")
# Retrieve the last timestamp if available
channel_messages = ui_state.all_messages[channel_id]
if channel_messages:
# Check the last entry for a timestamp
for entry in reversed(channel_messages):
if entry[0].startswith("--"):
last_hour = entry[0].strip("- ").strip()
break
else:
last_hour = None
else:
last_hour = None
# Add a new timestamp if it's a new hour
if last_hour != current_hour:
ui_state.all_messages[channel_id].append((f"-- {current_hour} --", ""))
ui_state.all_messages[channel_id].append((config.sent_message_prefix + config.ack_unknown_str + ": ", message))
add_new_message(channel_id, config.sent_message_prefix + config.ack_unknown_str + ": ", message)
timestamp = save_message_to_db(channel_id, myid, message)
@@ -230,10 +203,14 @@ def send_traceroute() -> None:
"""
Sends a RouteDiscovery protobuf to the selected node.
"""
channel_id = ui_state.node_list[ui_state.selected_node]
add_new_message(channel_id, f"{config.message_prefix} Sent Traceroute", "")
r = mesh_pb2.RouteDiscovery()
interface_state.interface.sendData(
r,
destinationId=ui_state.node_list[ui_state.selected_node],
destinationId=channel_id,
portNum=portnums_pb2.PortNum.TRACEROUTE_APP,
wantResponse=True,
onResponse=on_response_traceroute,

View File

@@ -1,6 +1,6 @@
import curses
import textwrap
import logging
import time
import traceback
from typing import Union
@@ -12,8 +12,51 @@ from contact.utilities.db_handler import get_name_from_database, update_node_inf
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
from contact.utilities.singleton import ui_state, interface_state
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 # set in main_ui
# 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()
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
@@ -23,25 +66,43 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
# Calculate window max dimensions
height, width = stdscr.getmaxyx()
# 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
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)
entry_height = 3
function_height = 3
y_pad = entry_height + function_height
packet_log_height = int(height / 3)
content_h = max(1, height - y_pad)
pkt_h = max(1, int(height / 3))
if firstrun:
entry_win = curses.newwin(entry_height, width, 0, 0)
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)
channel_win = curses.newwin(content_h, channel_width, entry_height, 0)
messages_win = curses.newwin(content_h, messages_width, entry_height, channel_width)
nodes_win = curses.newwin(content_h, nodes_width, entry_height, channel_width + messages_width)
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
)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - function_height, channel_width)
# Will be resized to what we need when drawn
messages_pad = curses.newpad(1, 1)
@@ -66,19 +127,19 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
entry_win.resize(3, width)
channel_win.resize(height - y_pad, channel_width)
channel_win.resize(content_h, channel_width)
messages_win.resize(height - y_pad, messages_width)
messages_win.resize(content_h, messages_width)
messages_win.mvwin(3, channel_width)
nodes_win.resize(height - y_pad, nodes_width)
nodes_win.resize(content_h, nodes_width)
nodes_win.mvwin(entry_height, channel_width + messages_width)
function_win.resize(3, width)
function_win.mvwin(height - function_height, 0)
packetlog_win.resize(packet_log_height, messages_width)
packetlog_win.mvwin(height - packet_log_height - function_height, channel_width)
packetlog_win.resize(pkt_h, messages_width)
packetlog_win.mvwin(height - pkt_h - function_height, channel_width)
# Draw window borders
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
@@ -93,6 +154,7 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
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.
@@ -103,13 +165,16 @@ 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"Input: {input_text[-(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()
@@ -138,7 +203,7 @@ def main_ui(stdscr: curses.window) -> None:
handle_leftright(char)
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
handle_enter(input_text)
input_text = handle_enter(input_text)
elif char == chr(20): # Ctrl + t for Traceroute
handle_ctrl_t(stdscr)
@@ -209,6 +274,8 @@ 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."""
@@ -220,29 +287,27 @@ 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 will bounds check for us
select_channel(ui_state.selected_channel - (channel_win.getmaxyx()[0] - 2))
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)) # select_node will bounds check for us
select_node(ui_state.selected_node - (nodes_win.getmaxyx()[0] - 2))
draw_window_arrows(ui_state.current_window)
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 will bounds check for us
select_channel(ui_state.selected_channel + (channel_win.getmaxyx()[0] - 2))
elif ui_state.current_window == 1:
msg_line_count = messages_pad.getmaxyx()[0]
ui_state.selected_message = min(
@@ -251,7 +316,8 @@ def handle_pagedown() -> None:
)
refresh_pad(1)
elif ui_state.current_window == 2:
select_node(ui_state.selected_node + (nodes_win.getmaxyx()[0] - 2)) # select_node will bounds check for us
select_node(ui_state.selected_node + (nodes_win.getmaxyx()[0] - 2))
draw_window_arrows(ui_state.current_window)
def handle_leftright(char: int) -> None:
@@ -259,46 +325,38 @@ 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:
channel_win.attrset(get_color("window_frame"))
channel_win.box()
channel_win.refresh()
paint_frame(channel_win, selected=False)
refresh_pad(0)
if old_window == 1:
messages_win.attrset(get_color("window_frame"))
messages_win.box()
messages_win.refresh()
paint_frame(messages_win, selected=False)
refresh_pad(1)
elif old_window == 2:
draw_function_win()
nodes_win.attrset(get_color("window_frame"))
nodes_win.box()
nodes_win.refresh()
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:
channel_win.attrset(get_color("window_frame_selected"))
channel_win.box()
channel_win.attrset(get_color("window_frame"))
channel_win.refresh()
paint_frame(channel_win, selected=True)
refresh_pad(0)
elif ui_state.current_window == 1:
messages_win.attrset(get_color("window_frame_selected"))
messages_win.box()
messages_win.attrset(get_color("window_frame"))
messages_win.refresh()
paint_frame(messages_win, selected=True)
refresh_pad(1)
elif ui_state.current_window == 2:
draw_function_win()
nodes_win.attrset(get_color("window_frame_selected"))
nodes_win.box()
nodes_win.attrset(get_color("window_frame"))
nodes_win.refresh()
paint_frame(nodes_win, selected=True)
refresh_pad(2)
# Draw arrows last; force even in multi-pane to avoid flicker
draw_window_arrows(ui_state.current_window)
def handle_enter(input_text: str) -> None:
def handle_enter(input_text: str) -> str:
"""Handle Enter key events to send messages or select channels."""
if ui_state.current_window == 2:
node_list = ui_state.node_list
@@ -315,34 +373,46 @@ def handle_enter(input_text: str) -> None:
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)
# Clear entry window and reset input text
input_text = ""
ui_state.last_sent_time = now
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_ctrl_t(stdscr: curses.window) -> None:
"""Handle Ctrl + T key events to send a traceroute."""
send_traceroute()
curses.curs_set(0) # Hide cursor
contact.ui.dialog.dialog(
stdscr,
"Traceroute Sent",
f"Traceroute Sent To: {get_name_from_database(ui_state.node_list[ui_state.selected_node])}",
"Results will appear in messages window.\nNote: Traceroute is limited to once every 30 seconds.",
)
curses.curs_set(1) # Show cursor again
handle_resize(stdscr, False)
def handle_backspace(entry_win: curses.window, input_text: str) -> None:
def handle_backspace(entry_win: curses.window, input_text: str) -> str:
"""Handle backspace key events to remove the last character from input text."""
if input_text:
input_text = input_text[:-1]
@@ -357,7 +427,10 @@ def handle_backspace(entry_win: curses.window, input_text: str) -> None:
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)
@@ -490,6 +563,10 @@ 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]
@@ -524,20 +601,18 @@ def draw_channel_list() -> None:
channel_pad.addstr(idx, 1, truncated_channel, color)
idx += 1
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()
paint_frame(channel_win, selected=(ui_state.current_window == 0))
refresh_pad(0)
draw_window_arrows(0)
channel_win.refresh()
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]
@@ -550,7 +625,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
row = 0
for prefix, message in messages:
full_message = f"{prefix}{message}"
wrapped_lines = textwrap.wrap(full_message, messages_win.getmaxyx()[1] - 2)
wrapped_lines = wrap_text(full_message, messages_win.getmaxyx()[1] - 2)
msg_line_count += len(wrapped_lines)
messages_pad.resize(msg_line_count, messages_win.getmaxyx()[1])
@@ -565,39 +640,32 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
messages_pad.addstr(row, 1, line, color)
row += 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()
paint_frame(messages_win, selected=(ui_state.current_window == 1))
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)
@@ -626,22 +694,20 @@ def draw_node_list() -> None:
i, 1, node_str, get_color(color, reverse=ui_state.selected_node == i and 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)
paint_frame(nodes_win, selected=(ui_state.current_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."""
@@ -699,15 +765,9 @@ 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:
@@ -744,6 +804,9 @@ 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()
@@ -782,9 +845,7 @@ def draw_packetlog_win() -> None:
# Add to the window
packetlog_win.addstr(i + 2, 1, logString, get_color("log"))
packetlog_win.attrset(get_color("window_frame"))
packetlog_win.box()
packetlog_win.refresh()
paint_frame(packetlog_win, selected=False)
# Restore cursor to input field
entry_win.keypad(True)
@@ -935,7 +996,7 @@ def draw_function_win() -> None:
def refresh_pad(window: int) -> None:
# Derive the target box and pad for the requested window
win_height = channel_win.getmaxyx()[0]
if window == 1:
@@ -963,13 +1024,43 @@ def refresh_pad(window: int) -> None:
selected_item = ui_state.selected_channel
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
if ui_state.single_pane_mode and window != ui_state.current_window:
return
# Compute inner drawable area of the box
box_y, box_x = box.getbegyx()
box_h, box_w = box.getmaxyx()
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,
box.getbegyx()[0] + 1,
box.getbegyx()[1] + 1,
box.getbegyx()[0] + lines,
box.getbegyx()[1] + box.getmaxyx()[1] - 3,
top,
left,
bottom,
right,
)

View File

@@ -6,6 +6,7 @@ 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 (
@@ -20,60 +21,72 @@ 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.ui.ui_state import MenuState
from contact.utilities.singleton import menu_state
menu_state = MenuState()
# Constants
width = 80
# Setup Variables
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
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.join(locals_dir, "node-configs")
config_folder = os.path.abspath(config.node_configs_file_path)
# Load translations
field_mapping, help_text = parse_ini_file(translation_file)
def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.window or pad types
def display_menu() -> tuple[object, object]:
if help_win:
min_help_window_height = 6
else:
min_help_window_height = 0
min_help_window_height = 6
num_items = len(menu_state.current_menu) + (1 if menu_state.show_save_option else 0)
# 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 - width) // 2
start_x = (curses.COLS - w) // 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, width, start_y, start_x)
menu_win = curses.newwin(menu_height, w, 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, width - 8)
menu_pad = curses.newpad(len(menu_state.current_menu) + 1, w - 8)
menu_pad.bkgd(get_color("background"))
header = " > ".join(word.title() for word in menu_state.menu_path)
if len(header) > width - 4:
header = header[: width - 7] + "..."
if len(header) > w - 4:
header = header[: w - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
transformed_path = transform_menu_path(menu_state.menu_path)
@@ -84,15 +97,15 @@ def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.wind
full_key = ".".join(transformed_path + [option])
display_name = field_mapping.get(full_key, option)
display_option = f"{display_name}"[: width // 2 - 2]
display_value = f"{current_value}"[: width // 2 - 4]
display_option = f"{display_name}"[: w // 2 - 2]
display_value = f"{current_value}"[: w // 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:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
menu_pad.addstr(idx, 0, f"{display_option:<{w // 2 - 2}} {display_value}".ljust(w - 8), color)
except curses.error:
pass
@@ -100,13 +113,13 @@ def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.wind
save_position = menu_height - 2
menu_win.addstr(
save_position,
(width - len(save_option)) // 2,
(w - 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, menu_state)
draw_help_window(start_y, start_x, menu_height, max_help_lines, transformed_path)
menu_win.refresh()
menu_pad.refresh(
@@ -117,6 +130,7 @@ def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.wind
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)
@@ -132,9 +146,7 @@ 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():
@@ -145,11 +157,21 @@ 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, width, help_y, menu_start_x
help_win, help_text, transformed_path, selected_option, max_help_lines, get_menu_width(), help_y, menu_start_x
)
def get_input_type_for_field(field) -> type:
if field.type in (field.TYPE_INT32, field.TYPE_UINT32, field.TYPE_INT64):
return int
elif field.type in (field.TYPE_FLOAT, field.TYPE_DOUBLE):
return float
else:
return str
def settings_menu(stdscr: object, interface: object) -> None:
curses.update_lines_cols()
@@ -159,29 +181,32 @@ def settings_menu(stdscr: object, interface: object) -> None:
modified_settings = {}
need_redraw = True
menu_state.need_redraw = True
menu_state.show_save_option = False
while True:
if need_redraw:
if menu_state.need_redraw:
menu_state.need_redraw = False
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(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)
(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)
)
# Display the menu
menu_win, menu_pad = display_menu(menu_state)
menu_win, menu_pad = display_menu()
need_redraw = False
if menu_win is None:
continue # Skip if menu_win is not initialized
# Capture user input
menu_win.timeout(200) # wait up to 200 ms for a keypress (or less if key is pressed)
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
@@ -215,14 +240,16 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
elif key == curses.KEY_RESIZE:
need_redraw = True
menu_state.need_redraw = True
curses.update_lines_cols()
menu_win.erase()
help_win.erase()
if help_win:
help_win.erase()
menu_win.refresh()
help_win.refresh()
if help_win:
help_win.refresh()
elif key == ord("\t") and menu_state.show_save_option:
old_selected_index = menu_state.selected_index
@@ -239,15 +266,17 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
elif key == curses.KEY_RIGHT or key == ord("\n"):
need_redraw = True
menu_state.need_redraw = True
menu_state.start_index.append(0)
menu_win.erase()
help_win.erase()
if help_win:
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()
help_win.refresh()
if help_win:
help_win.refresh()
if menu_state.show_save_option and menu_state.selected_index == len(options):
save_changes(interface, modified_settings, menu_state)
@@ -268,7 +297,8 @@ def settings_menu(stdscr: object, interface: object) -> None:
break
elif selected_option == "Export Config File":
filename = get_text_input("Enter a filename for the config file")
filename = get_text_input("Enter a filename for the config file", None, None)
if not filename:
logging.info("Export aborted: No filename provided.")
menu_state.start_index.pop()
@@ -290,7 +320,8 @@ def settings_menu(stdscr: object, interface: object) -> None:
with open(yaml_file_path, "w", encoding="utf-8") as file:
file.write(config_text)
logging.info(f"Config file saved to {yaml_file_path}")
dialog(stdscr, "Config File Saved:", yaml_file_path)
dialog("Config File Saved:", yaml_file_path)
menu_state.need_redraw = True
menu_state.start_index.pop()
continue
except PermissionError:
@@ -306,14 +337,16 @@ 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(stdscr, "", " No config files found. Export a config first.")
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))]
# Ensure file_list is not empty before proceeding
if not file_list:
dialog(stdscr, "", " No config files found. Export a config first.")
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)
@@ -327,7 +360,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
elif selected_option == "Config URL":
current_value = interface.localNode.getURL()
new_value = get_text_input(f"Config URL is currently: {current_value}")
new_value = get_text_input(f"Config URL is currently: {current_value}", None, str)
if new_value is not None:
current_value = new_value
overwrite = get_list_input(f"Are you sure you want to load this config?", None, ["Yes", "No"])
@@ -380,7 +413,6 @@ 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):
@@ -395,7 +427,9 @@ def settings_menu(stdscr: object, interface: object) -> None:
if selected_option in ["longName", "shortName", "isLicensed"]:
if selected_option in ["longName", "shortName"]:
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, None
)
new_value = current_value if new_value is None else new_value
menu_state.current_menu[selected_option] = (field, new_value)
@@ -414,7 +448,9 @@ def settings_menu(stdscr: object, interface: object) -> None:
menu_state.start_index.pop()
elif selected_option in ["latitude", "longitude", "altitude"]:
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, float
)
new_value = current_value if new_value is None else new_value
menu_state.current_menu[selected_option] = (field, new_value)
@@ -453,17 +489,26 @@ def settings_menu(stdscr: object, interface: object) -> None:
menu_state.start_index.pop()
elif field.type == 13: # Field type 13 corresponds to UINT32
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
input_type = get_input_type_for_field(field)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
)
new_value = current_value if new_value is None else int(new_value)
menu_state.start_index.pop()
elif field.type == 2: # Field type 13 corresponds to INT64
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
input_type = get_input_type_for_field(field)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
)
new_value = current_value if new_value is None else float(new_value)
menu_state.start_index.pop()
else: # Handle other field types
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
input_type = get_input_type_for_field(field)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
)
new_value = current_value if new_value is None else new_value
menu_state.start_index.pop()
@@ -486,19 +531,42 @@ def settings_menu(stdscr: object, interface: object) -> None:
menu_state.selected_index = 0
elif key == curses.KEY_LEFT:
need_redraw = True
# 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
menu_win.erase()
help_win.erase()
if help_win:
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()
help_win.refresh()
if help_win:
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:
@@ -515,6 +583,16 @@ 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,15 +2,69 @@ 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__))
parent_dir = os.path.abspath(os.path.join(script_dir, os.pardir))
# Paths
json_file_path = os.path.join(parent_dir, "config.json")
log_file_path = os.path.join(parent_dir, "client.log")
db_file_path = os.path.join(parent_dir, "client.db")
# To test writting to a non-writable directory, you can uncomment the following lines:
# mkdir /tmp/test_nonwritable
# chmod -w /tmp/test_nonwritable
# 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`.
"""
if not os.path.isdir(path):
return False
test_path = os.path.join(path, ".perm_test_tmp")
try:
with open(test_path, "w", encoding="utf-8") as _tmp:
_tmp.write("ok")
os.remove(test_path)
return True
except OSError:
return False
def _get_config_root(preferred_dir: str, fallback_name: str = ".contact_client") -> str:
"""
Choose a writable directory for config artifacts.
"""
if _is_writable_dir(preferred_dir):
return preferred_dir
home = os.path.expanduser("~")
fallback_dir = os.path.join(home, fallback_name)
# Ensure the fallback exists.
os.makedirs(fallback_dir, exist_ok=True)
# If *that* still isn't writable, last-ditch: use a system temp dir.
if not _is_writable_dir(fallback_dir):
import tempfile
fallback_dir = tempfile.mkdtemp(prefix="contact_client_")
return fallback_dir
# Pick the root now.
config_root = _get_config_root(parent_dir)
# Paths (derived from the chosen root)
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:
@@ -129,8 +183,10 @@ 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": "*",
@@ -171,16 +227,18 @@ 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, message_prefix, sent_message_prefix
global db_file_path, log_file_path, node_configs_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
global node_list_16ths, channel_list_16ths, single_pane_mode
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"]
@@ -212,6 +270,7 @@ 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,44 +1,63 @@
import curses
from contact.ui.colors import get_color
from contact.utilities.singleton import menu_state, ui_state
def dialog(stdscr: curses.window, title: str, message: str) -> None:
height, width = stdscr.getmaxyx()
def dialog(title: str, message: str) -> None:
"""Display a dialog with a title and message."""
# Calculate dialog dimensions
max_line_lengh = 0
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()
for l in message_lines:
max_line_length = max(len(l), max_line_lengh)
max_line_length = max(len(l) for l in message_lines)
dialog_width = max(len(title) + 4, max_line_length + 4)
dialog_height = len(message_lines) + 4
x = (width - dialog_width) // 2
y = (height - dialog_height) // 2
# Create dialog window
def draw_window():
win.erase()
win.bkgd(get_color("background"))
win.attrset(get_color("window_frame"))
win.border(0)
win.addstr(0, 2, title, get_color("settings_default"))
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"))
ok_text = " Ok "
win.addstr(
dialog_height - 2,
(dialog_width - len(ok_text)) // 2,
ok_text,
get_color("settings_default", reverse=True),
)
win.refresh()
win = curses.newwin(dialog_height, dialog_width, y, x)
win.bkgd(get_color("background"))
win.attrset(get_color("window_frame"))
win.border(0)
draw_window()
# Add title
win.addstr(0, 2, title, get_color("settings_default"))
# Add message
for i, l in enumerate(message_lines):
win.addstr(2 + i, 2, l, get_color("settings_default"))
# Add button
win.addstr(dialog_height - 2, (dialog_width - 4) // 2, " Ok ", get_color("settings_default", reverse=True))
# Refresh dialog window
win.refresh()
# Get user input
while True:
win.timeout(200)
char = win.getch()
# Close dialog with enter, space, or esc
if char in (curses.KEY_ENTER, 10, 13, 32, 27):
if menu_state.need_redraw:
menu_state.need_redraw = False
draw_window()
if char in (curses.KEY_ENTER, 10, 13, 32, 27): # Enter, space, Esc
win.erase()
win.refresh()
ui_state.current_window = previous_window
return
if char == -1:
continue

View File

@@ -1,5 +1,7 @@
import curses
import re
from unicodedata import east_asian_width
from contact.ui.colors import get_color
from contact.utilities.control_utils import transform_menu_path
from typing import Any, Optional, List, Dict
@@ -20,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(
@@ -71,9 +73,8 @@ def move_highlight(
# Clear old selection
if show_save_option and old_idx == max_index:
menu_win.chgat(
menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save")
)
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"))
else:
menu_pad.chgat(
old_idx,
@@ -88,9 +89,10 @@ 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(
menu_win.getmaxyx()[0] - 2,
(width - len(save_option)) // 2,
win_h - 2,
(win_w - len(save_option)) // 2,
len(save_option),
get_color("settings_save", reverse=True),
)
@@ -122,13 +124,14 @@ 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,
width,
win_w,
help_y,
menu_win.getbegyx()[1],
)
@@ -165,23 +168,46 @@ def update_help_window(
help_x: int,
) -> object: # returns a curses window
"""Handles rendering the help window consistently."""
wrapped_help = get_wrapped_help_text(help_text, transformed_path, selected_option, width, max_help_lines)
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)
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)
# Ensure help window does not exceed screen size
# Re-clamp Y to keep the window visible
if help_y + help_height > curses.LINES:
help_y = curses.LINES - help_height
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)
# Create or update the help window
if help_win is None:
help_win = curses.newwin(help_height, width, help_y, help_x)
help_win = curses.newwin(help_height, safe_width, help_y, help_x)
else:
help_win.erase()
help_win.refresh()
help_win.resize(help_height, width)
help_win.mvwin(help_y, help_x)
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.bkgd(get_color("background"))
help_win.attrset(get_color("window_frame"))
@@ -294,8 +320,17 @@ 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(" "))
text = text.translate(whitespace_trans)
words = re.findall(r"\S+|\s+", text) # Capture words and spaces separately
wrapped_lines = []
line_buffer = ""
@@ -304,11 +339,11 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
wrap_width -= margin
for word in words:
word_length = len(word)
word_length = text_width(word)
if word_length > wrap_width: # Break long words
if line_buffer:
wrapped_lines.append(line_buffer)
wrapped_lines.append(line_buffer.strip())
line_buffer = ""
line_length = 0
for i in range(0, word_length, wrap_width):
@@ -316,7 +351,7 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
continue
if line_length + word_length > wrap_width and word.strip():
wrapped_lines.append(line_buffer)
wrapped_lines.append(line_buffer.strip())
line_buffer = ""
line_length = 0
@@ -324,7 +359,7 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
line_length += word_length
if line_buffer:
wrapped_lines.append(line_buffer)
wrapped_lines.append(line_buffer.strip())
return wrapped_lines

View File

@@ -10,6 +10,7 @@ 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
@@ -24,11 +25,13 @@ class ChatUIState:
selected_message: int = 0
selected_node: int = 0
current_window: int = 0
last_sent_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,16 +4,23 @@ import curses
from typing import Any, List, Dict
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
from contact.ui.default_config import format_json_single_line_arrays, loaded_config
import contact.ui.default_config as 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
width = 80
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
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.
@@ -27,13 +34,14 @@ 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 = width - 16 # Allow space for "New Value: "
input_width = w - 16 # Allow space for "New Value: "
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
start_x = max(0, (curses.COLS - w) // 2)
# Create a centered window
edit_win = curses.newwin(height, width, start_y, start_x)
edit_win = curses.newwin(height, w, start_y, start_x)
edit_win.bkgd(get_color("background"))
edit_win.attrset(get_color("window_frame"))
edit_win.border()
@@ -42,7 +50,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 = width - 4 # Account for border and padding
wrap_width = w - 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
@@ -53,7 +61,9 @@ 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 loaded_config.keys() if k.startswith("COLOR_CONFIG")]
theme_options = [
k.split("_", 2)[2].lower() for k in config.loaded_config.keys() if k.startswith("COLOR_CONFIG")
]
return get_list_input("Select Theme", current_value, theme_options)
elif key == "node_sort":
@@ -64,6 +74,10 @@ 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)
@@ -72,41 +86,64 @@ 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:
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
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"))
edit_win.refresh()
edit_win.move(row, col + min(len(user_input) - scroll_offset, input_width)) # Adjust cursor position
key = edit_win.get_wch()
edit_win.move(row, col + min(len(user_input) - scroll_offset, input_width))
if key in (chr(27), curses.KEY_LEFT): # ESC or Left Arrow
try:
key = edit_win.get_wch()
except curses.error:
continue # window not ready — skip this loop
if key in (chr(27), curses.KEY_LEFT):
curses.curs_set(0)
return current_value # Exit without returning a value
return current_value
elif key in (chr(curses.KEY_ENTER), chr(10), chr(13)):
break
elif key in (curses.KEY_BACKSPACE, chr(127)): # Backspace
if user_input: # Only process if there's something to delete
elif key in (curses.KEY_BACKSPACE, chr(127)):
if user_input:
user_input = user_input[:-1]
if scroll_offset > 0 and len(user_input) < scroll_offset + input_width:
scroll_offset -= 1 # Move back if text is shorter than scrolled area
scroll_offset -= 1
else:
if isinstance(key, str):
user_input += key
else:
user_input += chr(key)
if len(user_input) > input_width: # Scroll if input exceeds visible area
if len(user_input) > input_width:
scroll_offset += 1
curses.curs_set(0)
return user_input if user_input else current_value
def display_menu(menu_state: Any) -> tuple[Any, Any, List[str]]:
def display_menu() -> tuple[Any, Any, List[str]]:
"""
Render the configuration menu with a Save button directly added to the window.
"""
@@ -124,11 +161,12 @@ def display_menu(menu_state: Any) -> 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 = (curses.COLS - width) // 2
start_x = max(0, (curses.COLS - w) // 2)
# Create the window
menu_win = curses.newwin(menu_height, width, start_y, start_x)
menu_win = curses.newwin(menu_height, w, start_y, start_x)
menu_win.erase()
menu_win.bkgd(get_color("background"))
menu_win.attrset(get_color("window_frame"))
@@ -136,13 +174,13 @@ def display_menu(menu_state: Any) -> tuple[Any, Any, List[str]]:
menu_win.keypad(True)
# Create the pad for scrolling
menu_pad = curses.newpad(num_items + 1, width - 8)
menu_pad = curses.newpad(num_items + 1, w - 8)
menu_pad.bkgd(get_color("background"))
# Display the menu path
header = " > ".join(menu_state.menu_path)
if len(header) > width - 4:
header = header[: width - 7] + "..."
if len(header) > w - 4:
header = header[: w - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
# Populate the pad with menu options
@@ -152,18 +190,18 @@ def display_menu(menu_state: Any) -> tuple[Any, Any, List[str]]:
if isinstance(menu_state.current_menu, dict)
else menu_state.current_menu[int(key.strip("[]"))]
)
display_key = f"{key}"[: width // 2 - 2]
display_value = f"{value}"[: width // 2 - 8]
display_key = f"{key}"[: w // 2 - 2]
display_value = f"{value}"[: w // 2 - 8]
color = get_color("settings_default", reverse=(idx == menu_state.selected_index))
menu_pad.addstr(idx, 0, f"{display_key:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
menu_pad.addstr(idx, 0, f"{display_key:<{w // 2 - 2}} {display_value}".ljust(w - 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,
(width - len(save_option)) // 2,
(w - len(save_option)) // 2,
save_option,
get_color("settings_save", reverse=(menu_state.selected_index == len(menu_state.current_menu))),
)
@@ -189,6 +227,7 @@ def display_menu(menu_state: Any) -> 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))
@@ -211,16 +250,18 @@ 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 need_redraw:
menu_win, menu_pad, options = display_menu(menu_state)
if menu_state.need_redraw:
menu_state.need_redraw = False
menu_win, menu_pad, options = display_menu()
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:
@@ -248,7 +289,7 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
elif key in (curses.KEY_RIGHT, 10, 13): # 10 = \n, 13 = carriage return
need_redraw = True
menu_state.need_redraw = True
menu_win.erase()
menu_win.refresh()
@@ -269,11 +310,14 @@ 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
@@ -282,22 +326,27 @@ 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
need_redraw = True
menu_state.need_redraw = True
if new_value != old:
made_changes = True
else:
# Save button selected
save_json(file_path, data)
made_changes = False
stdscr.refresh()
continue
# config.reload() # This isn't refreshing the file paths as expected
break
elif key in (27, curses.KEY_LEFT): # Escape or Left Arrow
need_redraw = True
menu_state.need_redraw = True
menu_win.erase()
menu_win.refresh()
@@ -318,6 +367,19 @@ 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()
@@ -325,7 +387,7 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
def save_json(file_path: str, data: Dict[str, Any]) -> None:
formatted_json = format_json_single_line_arrays(data)
formatted_json = config.format_json_single_line_arrays(data)
with open(file_path, "w", encoding="utf-8") as f:
f.write(formatted_json)
setup_colors(reinit=True)
@@ -334,7 +396,6 @@ 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

@@ -116,7 +116,14 @@ def load_messages_from_db() -> None:
# Add messages to ui_state.all_messages grouped by hourly timestamp
hourly_messages = {}
for user_id, message, timestamp, ack_type in db_messages:
for row in db_messages:
user_id, message, timestamp, ack_type = row
# Only ack_type is allowed to be None
if user_id is None or message is None or timestamp is None:
logging.warning(f"Skipping row with NULL required field(s): {row}")
continue
hour = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:00")
if hour not in hourly_messages:
hourly_messages[hour] = []
@@ -130,11 +137,13 @@ def load_messages_from_db() -> None:
ack_str = config.nak_str
if user_id == str(interface_state.myNodeNum):
formatted_message = (f"{config.sent_message_prefix}{ack_str}: ", message)
sanitized_message = message.replace("\x00", "")
formatted_message = (f"{config.sent_message_prefix}{ack_str}: ", sanitized_message)
else:
sanitized_message = message.replace("\x00", "")
formatted_message = (
f"{config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
message,
sanitized_message,
)
hourly_messages[hour].append(formatted_message)

View File

@@ -4,22 +4,71 @@ import curses
import ipaddress
from typing import Any, Optional, List
# 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
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
def get_text_input(prompt: str) -> Optional[str]:
def invalid_input(window: curses.window, message: str, redraw_func: Optional[callable] = None) -> None:
"""Displays an invalid input message in the given window and redraws if needed."""
cursor_y, cursor_x = window.getyx()
curses.curs_set(0)
dialog("Invalid Input", message)
if redraw_func:
redraw_func() # Redraw the original window content that got obscured
else:
window.refresh()
window.move(cursor_y, cursor_x)
curses.curs_set(1)
def get_text_input(prompt: str, selected_config: str, input_type: str) -> Optional[str]:
"""Handles user input with wrapped text for long prompts."""
def redraw_input_win():
"""Redraw the input window with the current prompt and user input."""
input_win.erase()
input_win.border()
row = 1
for line in wrapped_prompt:
input_win.addstr(row, margin, line[:input_width], get_color("settings_default", bold=True))
row += 1
if row >= height - 3:
break
input_win.addstr(row + 1, margin, prompt_text, get_color("settings_default"))
input_win.addstr(row + 1, col_start, user_input[:first_line_width], get_color("settings_default"))
for i, line in enumerate(wrap_text(user_input[first_line_width:], wrap_width=input_width)):
if row + 2 + i < height - 1:
input_win.addstr(row + 2 + i, margin, line[:input_width], get_color("settings_default"))
input_win.refresh()
height = 8
width = 80
width = get_dialog_width()
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 = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (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()
@@ -27,6 +76,7 @@ def get_text_input(prompt: str) -> Optional[str]:
# Wrap the prompt text
wrapped_prompt = wrap_text(prompt, wrap_width=input_width)
row = 1
for line in wrapped_prompt:
input_win.addstr(row, margin, line[:input_width], get_color("settings_default", bold=True))
row += 1
@@ -39,34 +89,125 @@ def get_text_input(prompt: str) -> Optional[str]:
input_win.refresh()
curses.curs_set(1)
max_length = 4 if "shortName" in prompt else None
user_input = ""
min_value = 0
max_value = 4294967295
min_length = 0
max_length = None
# Start user input after the prompt text
if selected_config is not None:
validation = get_validation_for(selected_config) or {}
min_value = validation.get("min_value", 0)
max_value = validation.get("max_value", 4294967295)
min_length = validation.get("min_length", 0)
max_length = validation.get("max_length")
user_input = ""
col_start = margin + len(prompt_text)
first_line_width = input_width - len(prompt_text) # Available space for first line
first_line_width = input_width - len(prompt_text)
while True:
key = input_win.get_wch() # Waits for user input
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw_input_win()
if key == chr(27) or key == curses.KEY_LEFT: # ESC or Left Arrow
try:
key = input_win.get_wch()
except curses.error:
continue
if key == chr(27) or key == curses.KEY_LEFT:
input_win.erase()
input_win.refresh()
curses.curs_set(0)
return None # Exit without saving
menu_state.need_redraw = True
return None
elif key in (chr(curses.KEY_ENTER), chr(10), chr(13)): # Enter key
break
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
length = len(user_input)
if min_length == max_length and max_length is not None:
if length != min_length:
invalid_input(
input_win, f"Value must be exactly {min_length} characters long.", redraw_func=redraw_input_win
)
continue
else:
if length < min_length:
invalid_input(
input_win,
f"Value must be at least {min_length} characters long.",
redraw_func=redraw_input_win,
)
continue
if max_length is not None and length > max_length:
invalid_input(
input_win,
f"Value must be no more than {max_length} characters long.",
redraw_func=redraw_input_win,
)
continue
if input_type is int:
if not user_input.isdigit():
invalid_input(input_win, "Only numeric digits (09) allowed.", redraw_func=redraw_input_win)
continue
int_val = int(user_input)
if not (min_value <= int_val <= max_value):
invalid_input(
input_win, f"Enter a number between {min_value} and {max_value}.", redraw_func=redraw_input_win
)
continue
curses.curs_set(0)
return int_val
elif input_type is float:
try:
float_val = float(user_input)
if not (min_value <= float_val <= max_value):
invalid_input(
input_win,
f"Enter a number between {min_value} and {max_value}.",
redraw_func=redraw_input_win,
)
continue
except ValueError:
invalid_input(input_win, "Must be a valid floating point number.", redraw_func=redraw_input_win)
continue
else:
curses.curs_set(0)
return float_val
else:
break
elif key in (curses.KEY_BACKSPACE, chr(127)): # Handle Backspace
if user_input:
user_input = user_input[:-1] # Remove last character
elif max_length is None or len(user_input) < max_length: # Enforce max length
if isinstance(key, str):
user_input += key
else:
user_input += chr(key)
elif max_length is None or len(user_input) < max_length:
try:
char = chr(key) if not isinstance(key, str) else key
if input_type is int:
if char.isdigit() or (char == "-" and len(user_input) == 0):
user_input += char
elif input_type is float:
if (
char.isdigit()
or (char == "." and "." not in user_input)
or (char == "-" and len(user_input) == 0)
):
user_input += char
else:
user_input += char
except ValueError:
pass # Ignore invalid input
# First line must be manually handled before using wrap_text()
first_line = user_input[:first_line_width] # Cut to max first line width
@@ -77,9 +218,7 @@ def get_text_input(prompt: str) -> Optional[str]:
# 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, " " * min(input_width, width - margin - 1), get_color("settings_default")
)
input_win.addstr(row + 1 + i, margin, " " * input_width, get_color("settings_default"))
# Redraw the prompt text so it never disappears
input_win.addstr(row + 1, margin, prompt_text, get_color("settings_default"))
@@ -95,10 +234,12 @@ def get_text_input(prompt: str) -> Optional[str]:
curses.curs_set(0)
input_win.erase()
input_win.refresh()
return user_input
return user_input.strip()
def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
"""Handles user input for editing up to 3 Admin Keys in Base64 format."""
def to_base64(byte_strings):
"""Convert byte values to Base64-encoded strings."""
return [base64.b64encode(b).decode() for b in byte_strings]
@@ -115,14 +256,15 @@ 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 = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (curses.COLS - width) // 2)
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
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
curses.echo()
curses.curs_set(1)
@@ -130,46 +272,48 @@ def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
# Editable list of values (max 3 values)
user_values = cvalue[:3] + [""] * (3 - len(cvalue)) # Ensure always 3 fields
cursor_pos = 0 # Track which value is being edited
error_message = ""
invalid_input = ""
while True:
repeated_win.erase()
repeated_win.border()
repeated_win.addstr(1, 2, "Edit up to 3 Admin Keys:", get_color("settings_default", bold=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))
# Display current values, allowing editing
for i, line in enumerate(user_values):
prefix = "" if i == cursor_pos else " " # Highlight the current line
repeated_win.addstr(
admin_key_win.addstr(
3 + i, 2, f"{prefix}Admin Key {i + 1}: ", get_color("settings_default", bold=(i == cursor_pos))
)
repeated_win.addstr(3 + i, 18, line) # Align text for easier editing
admin_key_win.addstr(3 + i, 18, line) # Align text for easier editing
# 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
admin_key_win.move(3 + cursor_pos, 18 + len(user_values[cursor_pos])) # Position cursor at end of text
# Show error message if needed
if error_message:
repeated_win.addstr(7, 2, error_message, get_color("settings_default", bold=True))
if invalid_input:
admin_key_win.addstr(7, 2, invalid_input, get_color("settings_default", bold=True))
repeated_win.refresh()
key = repeated_win.getch()
admin_key_win.refresh()
key = admin_key_win.getch()
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow -> Cancel and return original
repeated_win.erase()
repeated_win.refresh()
admin_key_win.erase()
admin_key_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)
return user_values # Return the edited Base64 values
else:
error_message = "Error: Each key must be valid Base64 and 32 bytes long!"
invalid_input = "Error: Each key must be valid Base64 and 32 bytes long!"
elif key == curses.KEY_UP: # Move cursor up
cursor_pos = (cursor_pos - 1) % len(user_values)
elif key == curses.KEY_DOWN: # Move cursor down
@@ -180,200 +324,252 @@ def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
else:
try:
user_values[cursor_pos] += chr(key) # Append valid character input to the selected field
error_message = "" # Clear error if user starts fixing input
invalid_input = "" # Clear error if user starts fixing input
except ValueError:
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 = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (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) # Enable keypad for special keys
repeated_win.keypad(True)
curses.echo()
curses.curs_set(1) # Show the cursor
curses.curs_set(1)
# Editable list of values (max 3 values)
user_values = current_value[:3]
cursor_pos = 0 # Track which value is being edited
error_message = ""
user_values = current_value[:3] + [""] * (3 - len(current_value)) # Always 3 fields
cursor_pos = 0
invalid_input = ""
while True:
def redraw():
repeated_win.erase()
repeated_win.border()
repeated_win.addstr(1, 2, "Edit up to 3 Values:", get_color("settings_default", bold=True))
# Display current values, allowing editing
win_h, win_w = repeated_win.getmaxyx()
for i, line in enumerate(user_values):
prefix = "" if i == cursor_pos else " " # Highlight the current line
prefix = "" if i == cursor_pos else " "
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)
repeated_win.addstr(3 + i, 18, line[: max(0, win_w - 20)]) # Prevent overflow
# 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 error_message:
repeated_win.addstr(7, 2, error_message, get_color("settings_default", bold=True))
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.move(3 + cursor_pos, 18 + len(user_values[cursor_pos]))
repeated_win.refresh()
key = repeated_win.getch()
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow -> Cancel and return original
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
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
elif key in ("\n", curses.KEY_ENTER):
curses.noecho()
curses.curs_set(0)
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
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]
else:
try:
user_values[cursor_pos] += chr(key) # Append valid character input to the selected field
error_message = "" # Clear error if user starts fixing input
except ValueError:
pass # Ignore invalid character inputs
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
def get_fixed32_input(current_value: int) -> int:
cvalue = current_value
current_value = str(ipaddress.IPv4Address(current_value))
original_value = current_value
ip_string = str(ipaddress.IPv4Address(current_value))
height = 10
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (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 = ""
while True:
def redraw():
fixed32_win.erase()
fixed32_win.border()
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.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.refresh()
key = fixed32_win.getch()
while True:
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw()
if key == 27 or key == curses.KEY_LEFT: # Escape or Left Arrow to cancel
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
fixed32_win.erase()
fixed32_win.refresh()
curses.noecho()
curses.curs_set(0)
return cvalue # Return the current value unchanged
elif key == ord("\n"): # Enter key to validate and save
# Validate IP address
menu_state.need_redraw = True
return original_value
elif key in ("\n", curses.KEY_ENTER):
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)
fixed32_address = ipaddress.ip_address(user_input)
return int(fixed32_address) # Return the valid IP address
return int(ipaddress.ip_address(user_input))
else:
fixed32_win.addstr(7, 2, "Invalid IP address. Try again.", curses.A_BOLD | curses.color_pair(5))
fixed32_win.addstr(7, 2, "Invalid IP address. Try again.", get_color("settings_default", bold=True))
fixed32_win.refresh()
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
curses.napms(1500)
user_input = ""
elif key in (curses.KEY_BACKSPACE, 127):
user_input = user_input[:-1]
else:
try:
char = chr(key)
if char.isdigit() or char == ".":
user_input += char # Append only valid characters (digits or dots)
except ValueError:
pass # Ignore invalid inputs
ch = chr(key) if isinstance(key, int) else key
if ch.isdigit() or ch == ".":
user_input += ch
except Exception:
pass # Ignore unprintable inputs
def get_list_input(prompt: str, current_option: Optional[str], list_options: List[str]) -> Optional[str]:
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]:
"""
Displays a scrollable list of list_options for the user to choose from.
List selector.
"""
selected_index = list_options.index(current_option) if current_option in list_options else 0
height = min(len(list_options) + 5, curses.LINES)
width = 80
start_y = (curses.LINES - height) // 2
start_x = (curses.COLS - width) // 2
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
start_x = max(0, (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, width - 8)
list_pad = curses.newpad(len(list_options) + 1, max(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
draw_arrows(list_win, visible_height, max_index, [0], show_save_option=False) # Initial call to draw arrows
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()
while True:
key = list_win.getch()
if menu_state.need_redraw:
menu_state.need_redraw = False
redraw_list_ui()
try:
key = list_win.getch()
except curses.error:
continue
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 key
elif key == ord("\n"): # Enter
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 Arrow
elif key == 27 or key == curses.KEY_LEFT: # ESC or Left
if mandatory:
continue
list_win.clear()
list_win.refresh()
menu_state.need_redraw = True
return current_option

View File

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

View File

@@ -1,4 +1,5 @@
import datetime
import time
from meshtastic.protobuf import config_pb2
import contact.ui.default_config as config
@@ -134,3 +135,31 @@ def get_time_ago(timestamp):
if unit != "s":
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] = []
# Timestamp handling
current_timestamp = time.time()
current_hour = datetime.datetime.fromtimestamp(current_timestamp).strftime("%Y-%m-%d %H:00")
# Retrieve the last timestamp if available
channel_messages = ui_state.all_messages[channel_id]
if channel_messages:
# Check the last entry for a timestamp
for entry in reversed(channel_messages):
if entry[0].startswith("--"):
last_hour = entry[0].strip("- ").strip()
break
else:
last_hour = None
else:
last_hour = None
# Add a new timestamp if it's a new hour
if last_hour != current_hour:
ui_state.all_messages[channel_id].append((f"-- {current_hour} --", ""))
# Add the message
ui_state.all_messages[channel_id].append((prefix,message))

View File

@@ -0,0 +1,23 @@
validation_rules = {
"shortName": {"max_length": 4},
"longName": {"max_length": 32},
"fixed_pin": {"min_length": 6, "max_length": 6},
"position_flags": {"max_length": 3},
"enabled_protocols": {"max_value": 2},
"hop_limit": {"max_value": 7},
"latitude": {"min_value": -90, "max_value": 90},
"longitude": {"min_value": -180, "max_value": 180},
"altitude": {"min_value": -4294967295, "max_value": 4294967295},
"red": {"max_value": 255},
"green": {"max_value": 255},
"blue": {"max_value": 255},
"current": {"max_value": 255},
"position_precision": {"max_value": 32},
}
def get_validation_for(key: str) -> dict:
for rule_key, config in validation_rules.items():
if rule_key in key:
return config
return {}

View File

@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.3.12"
version = "1.3.17"
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"}