forked from iarv/contact
Compare commits
36 Commits
notificati
...
errors
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
30402f4906 | ||
|
|
324e0b03e7 | ||
|
|
056db12911 | ||
|
|
685a2d4bf8 | ||
|
|
6ed0cc8c9f | ||
|
|
fc208a9258 | ||
|
|
eaf9381bca | ||
|
|
367af9044c | ||
|
|
d8183d9009 | ||
|
|
3fb1335be3 | ||
|
|
8b05072786 | ||
|
|
4455781e6c | ||
|
|
0c8aaee415 | ||
|
|
b97d9f4649 | ||
|
|
4152fb6a21 | ||
|
|
384e36dac2 | ||
|
|
65bca84fe6 | ||
|
|
16fa2830fd | ||
|
|
c8f1da99e3 | ||
|
|
702250c329 | ||
|
|
6291082405 | ||
|
|
4fa5148664 | ||
|
|
d62ec09eea | ||
|
|
61026dcc73 | ||
|
|
1362d3a219 | ||
|
|
981d72e688 | ||
|
|
0b5ec0b3d7 | ||
|
|
cbb4ef9e34 | ||
|
|
fecd71f4b7 | ||
|
|
59edfab451 | ||
|
|
39159099e1 | ||
|
|
02e5368c61 | ||
|
|
9d234a75d8 | ||
|
|
c7edd602ec | ||
|
|
00226c5b4d | ||
|
|
243079f8eb |
@@ -53,22 +53,21 @@ 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(args: object) -> 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()
|
||||
@@ -81,55 +80,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(args)
|
||||
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__":
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import logging
|
||||
import os
|
||||
import platform
|
||||
import time
|
||||
from datetime import datetime
|
||||
import shutil
|
||||
import subprocess
|
||||
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,
|
||||
@@ -25,12 +28,43 @@ from contact.utilities.singleton import ui_state, interface_state, app_state
|
||||
|
||||
|
||||
def play_sound():
|
||||
if platform.system() == "Darwin": # macOS
|
||||
os.system("afplay /System/Library/Sounds/Ping.aiff")
|
||||
elif platform.system() == "Linux":
|
||||
os.system("paplay /usr/share/sounds/freedesktop/stereo/complete.oga")
|
||||
else:
|
||||
print("\a") # fallback
|
||||
try:
|
||||
system = platform.system()
|
||||
sound_path = None
|
||||
executable = None
|
||||
|
||||
if system == "Darwin": # macOS
|
||||
sound_path = "/System/Library/Sounds/Ping.aiff"
|
||||
executable = "afplay"
|
||||
|
||||
elif system == "Linux":
|
||||
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("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}")
|
||||
|
||||
|
||||
def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
@@ -64,7 +98,9 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
maybe_store_nodeinfo_in_db(packet)
|
||||
|
||||
elif packet["decoded"]["portnum"] == "TEXT_MESSAGE_APP":
|
||||
play_sound()
|
||||
|
||||
if config.notification_sound == "True":
|
||||
play_sound()
|
||||
|
||||
message_bytes = packet["decoded"]["payload"]
|
||||
message_string = message_bytes.decode("utf-8")
|
||||
@@ -89,7 +125,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:
|
||||
@@ -99,40 +137,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}")
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import curses
|
||||
import textwrap
|
||||
import logging
|
||||
import traceback
|
||||
from typing import Union
|
||||
@@ -12,28 +11,36 @@ 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.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
|
||||
|
||||
|
||||
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
"""Handle terminal resize events and redraw the UI accordingly."""
|
||||
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, function_win, packetlog_win, entry_win
|
||||
|
||||
# Calculate window max dimensions
|
||||
height, width = stdscr.getmaxyx()
|
||||
|
||||
# Define window dimensions and positions
|
||||
channel_width = 3 * (width // 16)
|
||||
nodes_width = 5 * (width // 16)
|
||||
channel_width = int(config.channel_list_16ths) * (width // 16)
|
||||
nodes_width = int(config.node_list_16ths) * (width // 16)
|
||||
messages_width = width - channel_width - nodes_width
|
||||
|
||||
entry_height = 3
|
||||
function_height = 3
|
||||
y_pad = entry_height + function_height
|
||||
packet_log_height = int(height / 3)
|
||||
|
||||
if firstrun:
|
||||
entry_win = curses.newwin(3, width, 0, 0)
|
||||
channel_win = curses.newwin(height - 6, channel_width, 3, 0)
|
||||
messages_win = curses.newwin(height - 6, messages_width, 3, channel_width)
|
||||
nodes_win = curses.newwin(height - 6, nodes_width, 3, channel_width + messages_width)
|
||||
function_win = curses.newwin(3, width, height - 3, 0)
|
||||
packetlog_win = curses.newwin(int(height / 3), messages_width, height - int(height / 3) - 3, channel_width)
|
||||
entry_win = curses.newwin(entry_height, width, 0, 0)
|
||||
channel_win = curses.newwin(height - y_pad, channel_width, entry_height, 0)
|
||||
messages_win = curses.newwin(height - y_pad, messages_width, entry_height, channel_width)
|
||||
nodes_win = curses.newwin(height - y_pad, nodes_width, entry_height, channel_width + messages_width)
|
||||
function_win = curses.newwin(function_height, width, height - function_height, 0)
|
||||
packetlog_win = curses.newwin(
|
||||
packet_log_height, messages_width, height - packet_log_height - function_height, channel_width
|
||||
)
|
||||
|
||||
# Will be resized to what we need when drawn
|
||||
messages_pad = curses.newpad(1, 1)
|
||||
@@ -58,19 +65,19 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
|
||||
entry_win.resize(3, width)
|
||||
|
||||
channel_win.resize(height - 6, channel_width)
|
||||
channel_win.resize(height - y_pad, channel_width)
|
||||
|
||||
messages_win.resize(height - 6, messages_width)
|
||||
messages_win.resize(height - y_pad, messages_width)
|
||||
messages_win.mvwin(3, channel_width)
|
||||
|
||||
nodes_win.resize(height - 6, nodes_width)
|
||||
nodes_win.mvwin(3, channel_width + messages_width)
|
||||
nodes_win.resize(height - y_pad, nodes_width)
|
||||
nodes_win.mvwin(entry_height, channel_width + messages_width)
|
||||
|
||||
function_win.resize(3, width)
|
||||
function_win.mvwin(height - 3, 0)
|
||||
function_win.mvwin(height - function_height, 0)
|
||||
|
||||
packetlog_win.resize(int(height / 3), messages_width)
|
||||
packetlog_win.mvwin(height - int(height / 3) - 3, channel_width)
|
||||
packetlog_win.resize(packet_log_height, messages_width)
|
||||
packetlog_win.mvwin(height - packet_log_height - function_height, channel_width)
|
||||
|
||||
# Draw window borders
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
|
||||
@@ -93,6 +100,7 @@ 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
|
||||
input_text = ""
|
||||
stdscr.keypad(True)
|
||||
@@ -100,7 +108,7 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
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()
|
||||
@@ -108,307 +116,59 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
# draw_debug(f"Keypress: {char}")
|
||||
|
||||
if char == curses.KEY_UP:
|
||||
if ui_state.current_window == 0:
|
||||
scroll_channels(-1)
|
||||
elif ui_state.current_window == 1:
|
||||
scroll_messages(-1)
|
||||
elif ui_state.current_window == 2:
|
||||
scroll_nodes(-1)
|
||||
handle_up()
|
||||
|
||||
elif char == curses.KEY_DOWN:
|
||||
if ui_state.current_window == 0:
|
||||
scroll_channels(1)
|
||||
elif ui_state.current_window == 1:
|
||||
scroll_messages(1)
|
||||
elif ui_state.current_window == 2:
|
||||
scroll_nodes(1)
|
||||
handle_down()
|
||||
|
||||
elif char == curses.KEY_HOME:
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(0)
|
||||
elif ui_state.current_window == 1:
|
||||
ui_state.selected_message = 0
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
select_node(0)
|
||||
handle_home()
|
||||
|
||||
elif char == curses.KEY_END:
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(len(ui_state.channel_list) - 1)
|
||||
elif ui_state.current_window == 1:
|
||||
msg_line_count = messages_pad.getmaxyx()[0]
|
||||
ui_state.selected_message = max(msg_line_count - get_msg_window_lines(messages_win, packetlog_win), 0)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
select_node(len(ui_state.node_list) - 1)
|
||||
handle_end()
|
||||
|
||||
elif char == curses.KEY_PPAGE:
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(
|
||||
ui_state.selected_channel - (channel_win.getmaxyx()[0] - 2)
|
||||
) # select_channel will bounds check for us
|
||||
elif ui_state.current_window == 1:
|
||||
ui_state.selected_message = max(
|
||||
ui_state.selected_message - get_msg_window_lines(messages_win, packetlog_win), 0
|
||||
)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
select_node(
|
||||
ui_state.selected_node - (nodes_win.getmaxyx()[0] - 2)
|
||||
) # select_node will bounds check for us
|
||||
handle_pageup()
|
||||
|
||||
elif char == curses.KEY_NPAGE:
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(
|
||||
ui_state.selected_channel + (channel_win.getmaxyx()[0] - 2)
|
||||
) # select_channel will bounds check for us
|
||||
elif ui_state.current_window == 1:
|
||||
msg_line_count = messages_pad.getmaxyx()[0]
|
||||
ui_state.selected_message = min(
|
||||
ui_state.selected_message + get_msg_window_lines(messages_win, packetlog_win),
|
||||
msg_line_count - get_msg_window_lines(messages_win, packetlog_win),
|
||||
)
|
||||
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
|
||||
handle_pagedown()
|
||||
|
||||
elif char == curses.KEY_LEFT or char == curses.KEY_RIGHT:
|
||||
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
|
||||
|
||||
if old_window == 0:
|
||||
channel_win.attrset(get_color("window_frame"))
|
||||
channel_win.box()
|
||||
channel_win.refresh()
|
||||
refresh_pad(0)
|
||||
if old_window == 1:
|
||||
messages_win.attrset(get_color("window_frame"))
|
||||
messages_win.box()
|
||||
messages_win.refresh()
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
draw_function_win()
|
||||
nodes_win.attrset(get_color("window_frame"))
|
||||
nodes_win.box()
|
||||
nodes_win.refresh()
|
||||
refresh_pad(2)
|
||||
|
||||
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()
|
||||
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()
|
||||
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()
|
||||
refresh_pad(2)
|
||||
|
||||
# Check for Esc
|
||||
elif char == chr(27):
|
||||
break
|
||||
|
||||
# Check for Ctrl + t
|
||||
elif char == chr(20):
|
||||
send_traceroute()
|
||||
curses.curs_set(0) # Hide cursor
|
||||
contact.ui.dialog.dialog(
|
||||
stdscr,
|
||||
"Traceroute Sent",
|
||||
"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)
|
||||
handle_leftright(char)
|
||||
|
||||
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
|
||||
if ui_state.current_window == 2:
|
||||
node_list = ui_state.node_list
|
||||
if node_list[ui_state.selected_node] not in ui_state.channel_list:
|
||||
ui_state.channel_list.append(node_list[ui_state.selected_node])
|
||||
if node_list[ui_state.selected_node] not in ui_state.all_messages:
|
||||
ui_state.all_messages[node_list[ui_state.selected_node]] = []
|
||||
input_text = handle_enter(input_text)
|
||||
|
||||
ui_state.selected_channel = ui_state.channel_list.index(node_list[ui_state.selected_node])
|
||||
|
||||
if is_chat_archived(ui_state.channel_list[ui_state.selected_channel]):
|
||||
update_node_info_in_db(ui_state.channel_list[ui_state.selected_channel], chat_archived=False)
|
||||
|
||||
ui_state.selected_node = 0
|
||||
ui_state.current_window = 0
|
||||
|
||||
draw_node_list()
|
||||
draw_channel_list()
|
||||
draw_messages_window(True)
|
||||
|
||||
elif len(input_text) > 0:
|
||||
# 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 = ""
|
||||
entry_win.erase()
|
||||
elif char == chr(20): # Ctrl + t for Traceroute
|
||||
handle_ctrl_t(stdscr)
|
||||
|
||||
elif char in (curses.KEY_BACKSPACE, chr(127)):
|
||||
if input_text:
|
||||
input_text = input_text[:-1]
|
||||
y, x = entry_win.getyx()
|
||||
entry_win.move(y, x - 1)
|
||||
entry_win.addch(" ") #
|
||||
entry_win.move(y, x - 1)
|
||||
entry_win.refresh()
|
||||
input_text = handle_backspace(entry_win, input_text)
|
||||
|
||||
elif char == "`": # ` Launch the settings interface
|
||||
curses.curs_set(0)
|
||||
settings_menu(stdscr, interface_state.interface)
|
||||
curses.curs_set(1)
|
||||
refresh_node_list()
|
||||
handle_resize(stdscr, False)
|
||||
handle_backtick(stdscr)
|
||||
|
||||
elif char == chr(16):
|
||||
# Display packet log
|
||||
if ui_state.display_log is False:
|
||||
ui_state.display_log = True
|
||||
draw_messages_window(True)
|
||||
else:
|
||||
ui_state.display_log = False
|
||||
packetlog_win.erase()
|
||||
draw_messages_window(True)
|
||||
elif char == chr(16): # Ctrl + P for Packet Log
|
||||
handle_ctrl_p()
|
||||
|
||||
elif char == curses.KEY_RESIZE:
|
||||
input_text = ""
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
# ^D
|
||||
elif char == chr(4):
|
||||
if ui_state.current_window == 0:
|
||||
if isinstance(ui_state.channel_list[ui_state.selected_channel], int):
|
||||
update_node_info_in_db(ui_state.channel_list[ui_state.selected_channel], chat_archived=True)
|
||||
elif char == chr(4): # Ctrl + D to delete current channel or node
|
||||
handle_ctrl_d()
|
||||
|
||||
# Shift notifications up to account for deleted item
|
||||
for i in range(len(ui_state.notifications)):
|
||||
if ui_state.notifications[i] > ui_state.selected_channel:
|
||||
ui_state.notifications[i] -= 1
|
||||
elif char == chr(31): # Ctrl + / to search
|
||||
handle_ctrl_fslash()
|
||||
|
||||
del ui_state.channel_list[ui_state.selected_channel]
|
||||
ui_state.selected_channel = min(ui_state.selected_channel, len(ui_state.channel_list) - 1)
|
||||
select_channel(ui_state.selected_channel)
|
||||
draw_channel_list()
|
||||
draw_messages_window()
|
||||
elif char == chr(6): # Ctrl + F to toggle favorite
|
||||
handle_ctrl_f(stdscr)
|
||||
|
||||
if ui_state.current_window == 2:
|
||||
curses.curs_set(0)
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from nodedb?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeNode(ui_state.node_list[ui_state.selected_node])
|
||||
elif char == chr(7): # Ctrl + G to toggle ignored
|
||||
handle_ctlr_g(stdscr)
|
||||
|
||||
# Directly modifying the interface from client code - good? Bad? If it's stupid but it works, it's not supid?
|
||||
del interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
# Convert to "!hex" representation that interface.nodes uses
|
||||
hexid = f"!{hex(ui_state.node_list[ui_state.selected_node])[2:]}"
|
||||
del interface_state.interface.nodes[hexid]
|
||||
|
||||
ui_state.node_list.pop(ui_state.selected_node)
|
||||
|
||||
draw_messages_window()
|
||||
draw_node_list()
|
||||
else:
|
||||
draw_messages_window()
|
||||
curses.curs_set(1)
|
||||
continue
|
||||
|
||||
# ^/
|
||||
elif char == chr(31):
|
||||
if ui_state.current_window == 2 or ui_state.current_window == 0:
|
||||
search(ui_state.current_window)
|
||||
|
||||
# ^F
|
||||
elif char == chr(6):
|
||||
if ui_state.current_window == 2:
|
||||
selectedNode = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
curses.curs_set(0)
|
||||
|
||||
if "isFavorite" not in selectedNode or selectedNode["isFavorite"] == False:
|
||||
confirmation = get_list_input(
|
||||
f"Set {get_name_from_database(ui_state.node_list[ui_state.selected_node])} as Favorite?",
|
||||
None,
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.setFavorite(ui_state.node_list[ui_state.selected_node])
|
||||
# Maybe we shouldn't be modifying the nodedb, but maybe it should update itself
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]][
|
||||
"isFavorite"
|
||||
] = True
|
||||
|
||||
refresh_node_list()
|
||||
|
||||
else:
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from Favorites?",
|
||||
None,
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeFavorite(ui_state.node_list[ui_state.selected_node])
|
||||
# Maybe we shouldn't be modifying the nodedb, but maybe it should update itself
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]][
|
||||
"isFavorite"
|
||||
] = False
|
||||
|
||||
refresh_node_list()
|
||||
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
elif char == chr(7):
|
||||
if ui_state.current_window == 2:
|
||||
selectedNode = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
curses.curs_set(0)
|
||||
|
||||
if "isIgnored" not in selectedNode or selectedNode["isIgnored"] == False:
|
||||
confirmation = get_list_input(
|
||||
f"Set {get_name_from_database(ui_state.node_list[ui_state.selected_node])} as Ignored?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.setIgnored(ui_state.node_list[ui_state.selected_node])
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]][
|
||||
"isIgnored"
|
||||
] = True
|
||||
else:
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from Ignored?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeIgnored(ui_state.node_list[ui_state.selected_node])
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]][
|
||||
"isIgnored"
|
||||
] = False
|
||||
|
||||
handle_resize(stdscr, False)
|
||||
elif char == chr(27): # Escape to exit
|
||||
break
|
||||
|
||||
else:
|
||||
# Append typed character to input text
|
||||
@@ -418,7 +178,319 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
input_text += chr(char)
|
||||
|
||||
|
||||
def handle_up() -> None:
|
||||
"""Handle key up events to scroll the current window."""
|
||||
if ui_state.current_window == 0:
|
||||
scroll_channels(-1)
|
||||
elif ui_state.current_window == 1:
|
||||
scroll_messages(-1)
|
||||
elif ui_state.current_window == 2:
|
||||
scroll_nodes(-1)
|
||||
|
||||
|
||||
def handle_down() -> None:
|
||||
"""Handle key down events to scroll the current window."""
|
||||
if ui_state.current_window == 0:
|
||||
scroll_channels(1)
|
||||
elif ui_state.current_window == 1:
|
||||
scroll_messages(1)
|
||||
elif ui_state.current_window == 2:
|
||||
scroll_nodes(1)
|
||||
|
||||
|
||||
def handle_home() -> None:
|
||||
"""Handle home key events to select the first item in the current window."""
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(0)
|
||||
elif ui_state.current_window == 1:
|
||||
ui_state.selected_message = 0
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
select_node(0)
|
||||
|
||||
|
||||
def handle_end() -> None:
|
||||
"""Handle end key events to select the last item in the current window."""
|
||||
if ui_state.current_window == 0:
|
||||
select_channel(len(ui_state.channel_list) - 1)
|
||||
elif ui_state.current_window == 1:
|
||||
msg_line_count = messages_pad.getmaxyx()[0]
|
||||
ui_state.selected_message = max(msg_line_count - get_msg_window_lines(messages_win, packetlog_win), 0)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
select_node(len(ui_state.node_list) - 1)
|
||||
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
elif ui_state.current_window == 1:
|
||||
msg_line_count = messages_pad.getmaxyx()[0]
|
||||
ui_state.selected_message = min(
|
||||
ui_state.selected_message + get_msg_window_lines(messages_win, packetlog_win),
|
||||
msg_line_count - get_msg_window_lines(messages_win, packetlog_win),
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
def handle_leftright(char: int) -> None:
|
||||
"""Handle left/right key events to switch between windows."""
|
||||
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
|
||||
|
||||
if old_window == 0:
|
||||
channel_win.attrset(get_color("window_frame"))
|
||||
channel_win.box()
|
||||
channel_win.refresh()
|
||||
refresh_pad(0)
|
||||
if old_window == 1:
|
||||
messages_win.attrset(get_color("window_frame"))
|
||||
messages_win.box()
|
||||
messages_win.refresh()
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
draw_function_win()
|
||||
nodes_win.attrset(get_color("window_frame"))
|
||||
nodes_win.box()
|
||||
nodes_win.refresh()
|
||||
refresh_pad(2)
|
||||
|
||||
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()
|
||||
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()
|
||||
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()
|
||||
refresh_pad(2)
|
||||
|
||||
|
||||
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
|
||||
if node_list[ui_state.selected_node] not in ui_state.channel_list:
|
||||
ui_state.channel_list.append(node_list[ui_state.selected_node])
|
||||
if node_list[ui_state.selected_node] not in ui_state.all_messages:
|
||||
ui_state.all_messages[node_list[ui_state.selected_node]] = []
|
||||
|
||||
ui_state.selected_channel = ui_state.channel_list.index(node_list[ui_state.selected_node])
|
||||
|
||||
if is_chat_archived(ui_state.channel_list[ui_state.selected_channel]):
|
||||
update_node_info_in_db(ui_state.channel_list[ui_state.selected_channel], chat_archived=False)
|
||||
|
||||
ui_state.selected_node = 0
|
||||
ui_state.current_window = 0
|
||||
|
||||
draw_node_list()
|
||||
draw_channel_list()
|
||||
draw_messages_window(True)
|
||||
return input_text
|
||||
|
||||
elif len(input_text) > 0:
|
||||
# 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
|
||||
entry_win.erase()
|
||||
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,
|
||||
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) -> str:
|
||||
"""Handle backspace key events to remove the last character from input text."""
|
||||
if input_text:
|
||||
input_text = input_text[:-1]
|
||||
y, x = entry_win.getyx()
|
||||
entry_win.move(y, x - 1)
|
||||
entry_win.addch(" ") #
|
||||
entry_win.move(y, x - 1)
|
||||
entry_win.refresh()
|
||||
return input_text
|
||||
|
||||
|
||||
def handle_backtick(stdscr: curses.window) -> None:
|
||||
"""Handle backtick key events to open the settings menu."""
|
||||
curses.curs_set(0)
|
||||
settings_menu(stdscr, interface_state.interface)
|
||||
curses.curs_set(1)
|
||||
refresh_node_list()
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def handle_ctrl_p() -> None:
|
||||
"""Handle Ctrl + P key events to toggle the packet log display."""
|
||||
# Display packet log
|
||||
if ui_state.display_log is False:
|
||||
ui_state.display_log = True
|
||||
draw_messages_window(True)
|
||||
else:
|
||||
ui_state.display_log = False
|
||||
packetlog_win.erase()
|
||||
draw_messages_window(True)
|
||||
|
||||
|
||||
def handle_ctrl_d() -> None:
|
||||
if ui_state.current_window == 0:
|
||||
if isinstance(ui_state.channel_list[ui_state.selected_channel], int):
|
||||
update_node_info_in_db(ui_state.channel_list[ui_state.selected_channel], chat_archived=True)
|
||||
|
||||
# Shift notifications up to account for deleted item
|
||||
for i in range(len(ui_state.notifications)):
|
||||
if ui_state.notifications[i] > ui_state.selected_channel:
|
||||
ui_state.notifications[i] -= 1
|
||||
|
||||
del ui_state.channel_list[ui_state.selected_channel]
|
||||
ui_state.selected_channel = min(ui_state.selected_channel, len(ui_state.channel_list) - 1)
|
||||
select_channel(ui_state.selected_channel)
|
||||
draw_channel_list()
|
||||
draw_messages_window()
|
||||
|
||||
if ui_state.current_window == 2:
|
||||
curses.curs_set(0)
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from nodedb?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeNode(ui_state.node_list[ui_state.selected_node])
|
||||
|
||||
# Directly modifying the interface from client code - good? Bad? If it's stupid but it works, it's not supid?
|
||||
del interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
# Convert to "!hex" representation that interface.nodes uses
|
||||
hexid = f"!{hex(ui_state.node_list[ui_state.selected_node])[2:]}"
|
||||
del interface_state.interface.nodes[hexid]
|
||||
|
||||
ui_state.node_list.pop(ui_state.selected_node)
|
||||
|
||||
draw_messages_window()
|
||||
draw_node_list()
|
||||
else:
|
||||
draw_messages_window()
|
||||
curses.curs_set(1)
|
||||
|
||||
|
||||
def handle_ctrl_fslash() -> None:
|
||||
"""Handle Ctrl + / key events to search in the current window."""
|
||||
if ui_state.current_window == 2 or ui_state.current_window == 0:
|
||||
search(ui_state.current_window)
|
||||
|
||||
|
||||
def handle_ctrl_f(stdscr: curses.window) -> None:
|
||||
"""Handle Ctrl + F key events to toggle favorite status of the selected node."""
|
||||
if ui_state.current_window == 2:
|
||||
selectedNode = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
curses.curs_set(0)
|
||||
|
||||
if "isFavorite" not in selectedNode or selectedNode["isFavorite"] == False:
|
||||
confirmation = get_list_input(
|
||||
f"Set {get_name_from_database(ui_state.node_list[ui_state.selected_node])} as Favorite?",
|
||||
None,
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.setFavorite(ui_state.node_list[ui_state.selected_node])
|
||||
# Maybe we shouldn't be modifying the nodedb, but maybe it should update itself
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]["isFavorite"] = True
|
||||
|
||||
refresh_node_list()
|
||||
|
||||
else:
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from Favorites?",
|
||||
None,
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeFavorite(ui_state.node_list[ui_state.selected_node])
|
||||
# Maybe we shouldn't be modifying the nodedb, but maybe it should update itself
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]["isFavorite"] = False
|
||||
|
||||
refresh_node_list()
|
||||
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def handle_ctlr_g(stdscr: curses.window) -> None:
|
||||
"""Handle Ctrl + G key events to toggle ignored status of the selected node."""
|
||||
if ui_state.current_window == 2:
|
||||
selectedNode = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
curses.curs_set(0)
|
||||
|
||||
if "isIgnored" not in selectedNode or selectedNode["isIgnored"] == False:
|
||||
confirmation = get_list_input(
|
||||
f"Set {get_name_from_database(ui_state.node_list[ui_state.selected_node])} as Ignored?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.setIgnored(ui_state.node_list[ui_state.selected_node])
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]["isIgnored"] = True
|
||||
else:
|
||||
confirmation = get_list_input(
|
||||
f"Remove {get_name_from_database(ui_state.node_list[ui_state.selected_node])} from Ignored?",
|
||||
"No",
|
||||
["Yes", "No"],
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface_state.interface.localNode.removeIgnored(ui_state.node_list[ui_state.selected_node])
|
||||
interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]["isIgnored"] = False
|
||||
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def draw_channel_list() -> None:
|
||||
"""Update the channel list window and pad based on the current state."""
|
||||
channel_pad.erase()
|
||||
win_width = channel_win.getmaxyx()[1]
|
||||
|
||||
@@ -479,7 +551,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])
|
||||
|
||||
@@ -524,6 +596,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
|
||||
|
||||
|
||||
def draw_node_list() -> None:
|
||||
"""Update the nodes list window and pad based on the current state."""
|
||||
global nodes_pad
|
||||
|
||||
# This didn't work, for some reason an error is thown on startup, so we just create the pad every time
|
||||
@@ -572,6 +645,7 @@ def draw_node_list() -> None:
|
||||
|
||||
|
||||
def select_channel(idx: int) -> None:
|
||||
"""Select a channel by index and update the UI state accordingly."""
|
||||
old_selected_channel = ui_state.selected_channel
|
||||
ui_state.selected_channel = max(0, min(idx, len(ui_state.channel_list) - 1))
|
||||
draw_messages_window(True)
|
||||
@@ -593,6 +667,7 @@ def select_channel(idx: int) -> None:
|
||||
|
||||
|
||||
def scroll_channels(direction: int) -> None:
|
||||
"""Scroll through the channel list by a given direction."""
|
||||
new_selected_channel = ui_state.selected_channel + direction
|
||||
|
||||
if new_selected_channel < 0:
|
||||
@@ -604,6 +679,7 @@ def scroll_channels(direction: int) -> None:
|
||||
|
||||
|
||||
def scroll_messages(direction: int) -> None:
|
||||
"""Scroll through the messages in the current channel by a given direction."""
|
||||
ui_state.selected_message += direction
|
||||
|
||||
msg_line_count = messages_pad.getmaxyx()[0]
|
||||
@@ -636,6 +712,7 @@ def scroll_messages(direction: int) -> None:
|
||||
|
||||
|
||||
def select_node(idx: int) -> None:
|
||||
"""Select a node by index and update the UI state accordingly."""
|
||||
old_selected_node = ui_state.selected_node
|
||||
ui_state.selected_node = max(0, min(idx, len(ui_state.node_list) - 1))
|
||||
|
||||
@@ -652,6 +729,7 @@ def select_node(idx: int) -> None:
|
||||
|
||||
|
||||
def scroll_nodes(direction: int) -> None:
|
||||
"""Scroll through the node list by a given direction."""
|
||||
new_selected_node = ui_state.selected_node + direction
|
||||
|
||||
if new_selected_node < 0:
|
||||
@@ -663,7 +741,7 @@ def scroll_nodes(direction: int) -> None:
|
||||
|
||||
|
||||
def draw_packetlog_win() -> None:
|
||||
|
||||
"""Draw the packet log window with the latest packets."""
|
||||
columns = [10, 10, 15, 30]
|
||||
span = 0
|
||||
|
||||
@@ -716,6 +794,7 @@ def draw_packetlog_win() -> None:
|
||||
|
||||
|
||||
def search(win: int) -> None:
|
||||
"""Search for a node or channel based on user input."""
|
||||
start_idx = ui_state.selected_node
|
||||
select_func = select_node
|
||||
|
||||
@@ -764,6 +843,7 @@ def search(win: int) -> None:
|
||||
|
||||
|
||||
def draw_node_details() -> None:
|
||||
"""Draw the details of the selected node in the function window."""
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
@@ -827,16 +907,18 @@ def draw_node_details() -> None:
|
||||
|
||||
|
||||
def draw_help() -> None:
|
||||
"""Draw the help text in the function window."""
|
||||
cmds = [
|
||||
"↑→↓← = Select",
|
||||
" ENTER = Send",
|
||||
" ` = Settings",
|
||||
" ^P = Packet Log",
|
||||
" ESC = Quit",
|
||||
" ^t = Traceroute",
|
||||
" ^d = Archive Chat",
|
||||
" ^f = Favorite",
|
||||
" ^g = Ignore",
|
||||
" ENTER = Send",
|
||||
" ` = Settings",
|
||||
" ESC = Quit",
|
||||
" ^P = Packet Log",
|
||||
" ^t = Traceroute",
|
||||
" ^d = Archive Chat",
|
||||
" ^f = Favorite",
|
||||
" ^g = Ignore",
|
||||
" ^/ = Search",
|
||||
]
|
||||
function_str = ""
|
||||
for s in cmds:
|
||||
|
||||
@@ -123,15 +123,18 @@ def initialize_config() -> Dict[str, object]:
|
||||
"settings_breadcrumbs": ["green", "black"],
|
||||
"settings_warning": ["green", "black"],
|
||||
"settings_note": ["green", "black"],
|
||||
"node_favorite": ["cyan", "white"],
|
||||
"node_ignored": ["red", "white"],
|
||||
"node_favorite": ["cyan", "green"],
|
||||
"node_ignored": ["red", "black"],
|
||||
}
|
||||
default_config_variables = {
|
||||
"channel_list_16ths": "3",
|
||||
"node_list_16ths": "5",
|
||||
"db_file_path": db_file_path,
|
||||
"log_file_path": log_file_path,
|
||||
"message_prefix": ">>",
|
||||
"sent_message_prefix": ">> Sent",
|
||||
"notification_symbol": "*",
|
||||
"notification_sound": "True",
|
||||
"ack_implicit_str": "[◌]",
|
||||
"ack_str": "[✓]",
|
||||
"nak_str": "[x]",
|
||||
@@ -170,18 +173,23 @@ def assign_config_variables(loaded_config: Dict[str, object]) -> None:
|
||||
|
||||
global db_file_path, log_file_path, message_prefix, sent_message_prefix
|
||||
global notification_symbol, ack_implicit_str, ack_str, nak_str, ack_unknown_str
|
||||
global node_list_16ths, channel_list_16ths
|
||||
global theme, COLOR_CONFIG
|
||||
global node_sort
|
||||
global node_sort, notification_sound
|
||||
|
||||
channel_list_16ths = loaded_config["channel_list_16ths"]
|
||||
node_list_16ths = loaded_config["node_list_16ths"]
|
||||
db_file_path = loaded_config["db_file_path"]
|
||||
log_file_path = loaded_config["log_file_path"]
|
||||
message_prefix = loaded_config["message_prefix"]
|
||||
sent_message_prefix = loaded_config["sent_message_prefix"]
|
||||
notification_symbol = loaded_config["notification_symbol"]
|
||||
notification_sound = loaded_config["notification_sound"]
|
||||
ack_implicit_str = loaded_config["ack_implicit_str"]
|
||||
ack_str = loaded_config["ack_str"]
|
||||
nak_str = loaded_config["nak_str"]
|
||||
ack_unknown_str = loaded_config["ack_unknown_str"]
|
||||
node_sort = loaded_config["node_sort"]
|
||||
theme = loaded_config["theme"]
|
||||
if theme == "dark":
|
||||
COLOR_CONFIG = loaded_config["COLOR_CONFIG_DARK"]
|
||||
@@ -189,7 +197,6 @@ def assign_config_variables(loaded_config: Dict[str, object]) -> None:
|
||||
COLOR_CONFIG = loaded_config["COLOR_CONFIG_LIGHT"]
|
||||
elif theme == "green":
|
||||
COLOR_CONFIG = loaded_config["COLOR_CONFIG_GREEN"]
|
||||
node_sort = loaded_config["node_sort"]
|
||||
|
||||
|
||||
# Call the function when the script is imported
|
||||
|
||||
@@ -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
|
||||
@@ -293,9 +295,16 @@ 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 +313,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 +325,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 +333,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
|
||||
|
||||
|
||||
@@ -55,10 +55,15 @@ def edit_value(key: str, current_value: str) -> str:
|
||||
# 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")]
|
||||
return get_list_input("Select Theme", current_value, theme_options)
|
||||
|
||||
elif key == "node_sort":
|
||||
sort_options = ["lastHeard", "name", "hops"]
|
||||
return get_list_input("Sort By", current_value, sort_options)
|
||||
|
||||
elif key == "notification_sound":
|
||||
sound_options = ["True", "False"]
|
||||
return get_list_input("Notification Sound", current_value, sound_options)
|
||||
|
||||
# Standard Input Mode (Scrollable)
|
||||
edit_win.addstr(7, 2, "New Value: ", get_color("settings_default"))
|
||||
curses.curs_set(1)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import yaml
|
||||
import logging
|
||||
import time
|
||||
from typing import List
|
||||
from google.protobuf.json_format import MessageToDict
|
||||
from meshtastic import mt_config
|
||||
@@ -133,24 +134,29 @@ def config_import(interface, filename):
|
||||
logging.info(f"Setting device owner to {configuration['owner']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(configuration["owner"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "owner_short" in configuration:
|
||||
logging.info(f"Setting device owner short to {configuration['owner_short']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["owner_short"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "ownerShort" in configuration:
|
||||
logging.info(f"Setting device owner short to {configuration['ownerShort']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["ownerShort"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "channel_url" in configuration:
|
||||
logging.info(f"Setting channel url to {configuration['channel_url']}")
|
||||
interface.getNode("^local").setURL(configuration["channel_url"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "channelUrl" in configuration:
|
||||
logging.info(f"Setting channel url to {configuration['channelUrl']}")
|
||||
interface.getNode("^local").setURL(configuration["channelUrl"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "location" in configuration:
|
||||
alt = 0
|
||||
@@ -169,12 +175,14 @@ def config_import(interface, filename):
|
||||
logging.info(f"Fixing longitude at {lon} degrees")
|
||||
logging.info("Setting device position")
|
||||
interface.localNode.setFixedPosition(lat, lon, alt)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "config" in configuration:
|
||||
localConfig = interface.getNode("^local").localConfig
|
||||
for section in configuration["config"]:
|
||||
traverseConfig(section, configuration["config"][section], localConfig)
|
||||
interface.getNode("^local").writeConfig(camel_to_snake(section))
|
||||
time.sleep(0.5)
|
||||
|
||||
if "module_config" in configuration:
|
||||
moduleConfig = interface.getNode("^local").moduleConfig
|
||||
@@ -185,6 +193,7 @@ def config_import(interface, filename):
|
||||
moduleConfig,
|
||||
)
|
||||
interface.getNode("^local").writeConfig(camel_to_snake(section))
|
||||
time.sleep(0.5)
|
||||
|
||||
interface.getNode("^local", False).commitSettingsTransaction()
|
||||
logging.info("Writing modified configuration to device")
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.3.10"
|
||||
version = "1.3.15"
|
||||
description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
|
||||
authors = [
|
||||
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
|
||||
|
||||
Reference in New Issue
Block a user