mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
Compare commits
10 Commits
1.4.21
...
fix-search
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b43e3f4868 | ||
|
|
2e8e21f5ba | ||
|
|
09f0d626df | ||
|
|
8c37f2394b | ||
|
|
568f29ee29 | ||
|
|
87127adef1 | ||
|
|
dd0eb1473c | ||
|
|
7d9703a548 | ||
|
|
68b8d15181 | ||
|
|
4ef87871df |
@@ -77,7 +77,7 @@ help.node_info, "F5 = Full node info", ""
|
||||
help.archive_chat, "Ctrl+D = Archive chat / remove node", ""
|
||||
help.favorite, "Ctrl+F = Favorite", ""
|
||||
help.ignore, "Ctrl+G = Ignore", ""
|
||||
help.search, "Ctrl+/ = Search", ""
|
||||
help.search, "Ctrl+/ or / = Search", ""
|
||||
help.help, "Ctrl+K = Help", ""
|
||||
help.no_help, "No help available.", ""
|
||||
confirm.remove_from_nodedb, "Remove {name} from nodedb?", ""
|
||||
|
||||
@@ -77,7 +77,7 @@ help.node_info, "F5 = Полная информация об узле", ""
|
||||
help.archive_chat, "Ctrl+D = Архив чата / удалить узел", ""
|
||||
help.favorite, "Ctrl+F = Избранное", ""
|
||||
help.ignore, "Ctrl+G = Игнорировать", ""
|
||||
help.search, "Ctrl+/ = Поиск", ""
|
||||
help.search, "Ctrl+/ или / = Поиск", ""
|
||||
help.help, "Ctrl+K = Справка", ""
|
||||
help.no_help, "Нет справки.", ""
|
||||
confirm.remove_from_nodedb, "Удалить {name} из базы узлов?", ""
|
||||
|
||||
@@ -12,15 +12,17 @@ from contact.ui.colors import get_color
|
||||
from contact.utilities.db_handler import get_name_from_database, update_node_info_in_db, is_chat_archived
|
||||
from contact.utilities.input_handlers import get_list_input
|
||||
from contact.utilities.i18n import t
|
||||
from contact.utilities.emoji_utils import normalize_message_text
|
||||
import contact.ui.default_config as config
|
||||
import contact.ui.dialog
|
||||
from contact.ui.nav_utils import draw_main_arrows, fit_text, get_msg_window_lines, move_main_highlight, wrap_text
|
||||
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
|
||||
RESIZE_DEBOUNCE_MS = 250
|
||||
root_win = None
|
||||
nodes_pad = None
|
||||
|
||||
|
||||
# Draw arrows for a specific window id (0=channel,1=messages,2=nodes).
|
||||
@@ -63,6 +65,72 @@ def paint_frame(win, selected: bool) -> None:
|
||||
win.refresh()
|
||||
|
||||
|
||||
def get_channel_row_color(index: int) -> int:
|
||||
if index == ui_state.selected_channel:
|
||||
if ui_state.current_window == 0:
|
||||
return get_color("channel_list", reverse=True)
|
||||
return get_color("channel_selected")
|
||||
return get_color("channel_list")
|
||||
|
||||
|
||||
def get_node_row_color(index: int, highlight: bool = False) -> int:
|
||||
node_num = ui_state.node_list[index]
|
||||
node = interface_state.interface.nodesByNum.get(node_num, {})
|
||||
color = "node_list"
|
||||
if node.get("isFavorite"):
|
||||
color = "node_favorite"
|
||||
if node.get("isIgnored"):
|
||||
color = "node_ignored"
|
||||
reverse = index == ui_state.selected_node and (ui_state.current_window == 2 or highlight)
|
||||
return get_color(color, reverse=reverse)
|
||||
|
||||
|
||||
def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None:
|
||||
if nodes_pad is None or not ui_state.node_list:
|
||||
return
|
||||
|
||||
width = max(0, nodes_pad.getmaxyx()[1] - 4)
|
||||
|
||||
if 0 <= old_index < len(ui_state.node_list):
|
||||
try:
|
||||
nodes_pad.chgat(old_index, 1, width, get_node_row_color(old_index, highlight=highlight))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
if 0 <= ui_state.selected_node < len(ui_state.node_list):
|
||||
try:
|
||||
nodes_pad.chgat(ui_state.selected_node, 1, width, get_node_row_color(ui_state.selected_node, highlight=highlight))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
ui_state.start_index[2] = max(0, ui_state.selected_node - (nodes_win.getmaxyx()[0] - 3))
|
||||
refresh_pad(2)
|
||||
draw_window_arrows(2)
|
||||
|
||||
|
||||
def refresh_main_window(window_id: int, selected: bool) -> None:
|
||||
if window_id == 0:
|
||||
paint_frame(channel_win, selected=selected)
|
||||
if ui_state.channel_list:
|
||||
width = max(0, channel_pad.getmaxyx()[1] - 4)
|
||||
channel_pad.chgat(ui_state.selected_channel, 1, width, get_channel_row_color(ui_state.selected_channel))
|
||||
refresh_pad(0)
|
||||
elif window_id == 1:
|
||||
paint_frame(messages_win, selected=selected)
|
||||
refresh_pad(1)
|
||||
elif window_id == 2:
|
||||
paint_frame(nodes_win, selected=selected)
|
||||
if ui_state.node_list and nodes_pad is not None:
|
||||
width = max(0, nodes_pad.getmaxyx()[1] - 4)
|
||||
nodes_pad.chgat(ui_state.selected_node, 1, width, get_node_row_color(ui_state.selected_node))
|
||||
refresh_pad(2)
|
||||
|
||||
|
||||
def get_node_display_name(node_num: int, node: dict) -> str:
|
||||
user = node.get("user") or {}
|
||||
return user.get("longName") or get_name_from_database(node_num, "long")
|
||||
|
||||
|
||||
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
"""Handle terminal resize events and redraw the UI accordingly."""
|
||||
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, packetlog_win, entry_win
|
||||
@@ -255,7 +323,9 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
elif char == chr(4): # Ctrl + D to delete current channel or node
|
||||
handle_ctrl_d()
|
||||
|
||||
elif char == chr(31): # Ctrl + / to search
|
||||
elif char == chr(31) or (
|
||||
char == "/" and not input_text and ui_state.current_window in (0, 2)
|
||||
): # Ctrl + / or / to search in channel/node lists
|
||||
handle_ctrl_fslash()
|
||||
|
||||
elif char == chr(11): # Ctrl + K for Help
|
||||
@@ -359,31 +429,16 @@ 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 ui_state.single_pane_mode:
|
||||
handle_resize(root_win, False)
|
||||
return
|
||||
|
||||
if old_window == 0:
|
||||
paint_frame(channel_win, selected=False)
|
||||
refresh_pad(0)
|
||||
if old_window == 1:
|
||||
paint_frame(messages_win, selected=False)
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
paint_frame(nodes_win, selected=False)
|
||||
refresh_pad(2)
|
||||
refresh_main_window(old_window, selected=False)
|
||||
|
||||
if not ui_state.single_pane_mode:
|
||||
draw_window_arrows(old_window)
|
||||
|
||||
if ui_state.current_window == 0:
|
||||
paint_frame(channel_win, selected=True)
|
||||
refresh_pad(0)
|
||||
elif ui_state.current_window == 1:
|
||||
paint_frame(messages_win, selected=True)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
paint_frame(nodes_win, selected=True)
|
||||
refresh_pad(2)
|
||||
|
||||
refresh_main_window(ui_state.current_window, selected=True)
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
|
||||
@@ -404,31 +459,16 @@ def handle_function_keys(char: int) -> None:
|
||||
return
|
||||
|
||||
ui_state.current_window = target
|
||||
handle_resize(root_win, False)
|
||||
if ui_state.single_pane_mode:
|
||||
handle_resize(root_win, False)
|
||||
return
|
||||
|
||||
if old_window == 0:
|
||||
paint_frame(channel_win, selected=False)
|
||||
refresh_pad(0)
|
||||
elif old_window == 1:
|
||||
paint_frame(messages_win, selected=False)
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
paint_frame(nodes_win, selected=False)
|
||||
refresh_pad(2)
|
||||
refresh_main_window(old_window, selected=False)
|
||||
|
||||
if not ui_state.single_pane_mode:
|
||||
draw_window_arrows(old_window)
|
||||
|
||||
if ui_state.current_window == 0:
|
||||
paint_frame(channel_win, selected=True)
|
||||
refresh_pad(0)
|
||||
elif ui_state.current_window == 1:
|
||||
paint_frame(messages_win, selected=True)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
paint_frame(nodes_win, selected=True)
|
||||
refresh_pad(2)
|
||||
|
||||
refresh_main_window(ui_state.current_window, selected=True)
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
|
||||
@@ -480,34 +520,34 @@ def handle_enter(input_text: str) -> str:
|
||||
|
||||
|
||||
def handle_f5_key(stdscr: curses.window) -> None:
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
if not ui_state.node_list:
|
||||
return
|
||||
|
||||
def build_node_details() -> tuple[str, list[str]]:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
message_parts = []
|
||||
|
||||
message_parts.append("**📋 Basic Information:**")
|
||||
message_parts.append(f"• Device: {node.get('user', {}).get('longName', 'Unknown')}")
|
||||
message_parts.append(f"• Short name: {node.get('user', {}).get('shortName', 'Unknown')}")
|
||||
message_parts.append(f"• Hardware: {node.get('user', {}).get('hwModel', 'Unknown')}")
|
||||
|
||||
role = f"{node.get('user', {}).get('role', 'Unknown')}"
|
||||
message_parts.append(f"• Role: {role}")
|
||||
|
||||
pk = f"{node.get('user', {}).get('publicKey')}"
|
||||
message_parts.append(f"Public key: {pk}")
|
||||
|
||||
message_parts.append(f"• Role: {node.get('user', {}).get('role', 'Unknown')}")
|
||||
message_parts.append(f"Public key: {node.get('user', {}).get('publicKey')}")
|
||||
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
|
||||
|
||||
if "position" in node:
|
||||
pos = node["position"]
|
||||
if pos.get("latitude") and pos.get("longitude"):
|
||||
has_coords = pos.get("latitude") and pos.get("longitude")
|
||||
if has_coords:
|
||||
message_parts.append(f"• Position: {pos['latitude']:.4f}, {pos['longitude']:.4f}")
|
||||
if pos.get("altitude"):
|
||||
message_parts.append(f"• Altitude: {pos['altitude']}m")
|
||||
message_parts.append(f"https://maps.google.com/?q={pos['latitude']:.4f},{pos['longitude']:.4f}")
|
||||
if has_coords:
|
||||
message_parts.append(f"https://maps.google.com/?q={pos['latitude']:.4f},{pos['longitude']:.4f}")
|
||||
|
||||
if any(key in node for key in ["snr", "hopsAway", "lastHeard"]):
|
||||
message_parts.append("\n**🌐 Network Metrics:**")
|
||||
message_parts.append("")
|
||||
message_parts.append("**🌐 Network Metrics:**")
|
||||
|
||||
if "snr" in node:
|
||||
snr = node["snr"]
|
||||
@@ -527,12 +567,13 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
hop_emoji = "📡" if hops == 0 else "🔄" if hops == 1 else "⏩"
|
||||
message_parts.append(f"• Hops away: {hop_emoji} {hops}")
|
||||
|
||||
if "lastHeard" in node and node["lastHeard"]:
|
||||
if node.get("lastHeard"):
|
||||
message_parts.append(f"• Last heard: 🕐 {get_time_ago(node['lastHeard'])}")
|
||||
|
||||
if node.get("deviceMetrics"):
|
||||
metrics = node["deviceMetrics"]
|
||||
message_parts.append("\n**📊 Device Metrics:**")
|
||||
message_parts.append("")
|
||||
message_parts.append("**📊 Device Metrics:**")
|
||||
|
||||
if "batteryLevel" in metrics:
|
||||
battery = metrics["batteryLevel"]
|
||||
@@ -553,21 +594,128 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
air_emoji = "🔴" if air_util > 80 else "🟡" if air_util > 50 else "🟢"
|
||||
message_parts.append(f"• Air utilization TX: {air_emoji} {air_util:.2f}%")
|
||||
|
||||
message = "\n".join(message_parts)
|
||||
|
||||
contact.ui.dialog.dialog(
|
||||
t(
|
||||
"ui.dialog.node_details_title",
|
||||
default="📡 Node Details: {name}",
|
||||
name=node.get("user", {}).get("shortName", "Unknown"),
|
||||
),
|
||||
message,
|
||||
title = t(
|
||||
"ui.dialog.node_details_title",
|
||||
default="📡 Node Details: {name}",
|
||||
name=node.get("user", {}).get("shortName", "Unknown"),
|
||||
)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
return title, message_parts
|
||||
|
||||
previous_window = ui_state.current_window
|
||||
ui_state.current_window = 4
|
||||
scroll_offset = 0
|
||||
dialog_win = None
|
||||
|
||||
curses.curs_set(0)
|
||||
refresh_node_selection(highlight=True)
|
||||
|
||||
try:
|
||||
while True:
|
||||
curses.update_lines_cols()
|
||||
height, width = curses.LINES, curses.COLS
|
||||
title, message_lines = build_node_details()
|
||||
|
||||
max_line_length = max(len(title), *(len(line) for line in message_lines))
|
||||
dialog_width = min(max(max_line_length + 4, 20), max(10, width - 2))
|
||||
dialog_height = min(max(len(message_lines) + 4, 6), max(6, height - 2))
|
||||
x = max(0, (width - dialog_width) // 2)
|
||||
y = max(0, (height - dialog_height) // 2)
|
||||
viewport_h = max(1, dialog_height - 4)
|
||||
max_scroll = max(0, len(message_lines) - viewport_h)
|
||||
scroll_offset = max(0, min(scroll_offset, max_scroll))
|
||||
|
||||
if dialog_win is None:
|
||||
dialog_win = curses.newwin(dialog_height, dialog_width, y, x)
|
||||
else:
|
||||
dialog_win.erase()
|
||||
dialog_win.refresh()
|
||||
dialog_win.resize(dialog_height, dialog_width)
|
||||
dialog_win.mvwin(y, x)
|
||||
|
||||
dialog_win.keypad(True)
|
||||
dialog_win.bkgd(get_color("background"))
|
||||
dialog_win.attrset(get_color("window_frame"))
|
||||
dialog_win.border(0)
|
||||
|
||||
try:
|
||||
dialog_win.addstr(0, 2, title[: max(0, dialog_width - 4)], get_color("settings_default"))
|
||||
hint = f" {ui_state.selected_node + 1}/{len(ui_state.node_list)} "
|
||||
dialog_win.addstr(0, max(2, dialog_width - len(hint) - 2), hint, get_color("commands"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
msg_win = dialog_win.derwin(viewport_h + 2, dialog_width - 2, 1, 1)
|
||||
msg_win.erase()
|
||||
|
||||
for row, line in enumerate(message_lines[scroll_offset : scroll_offset + viewport_h], start=1):
|
||||
trimmed = line[: max(0, dialog_width - 6)]
|
||||
try:
|
||||
msg_win.addstr(row, 1, trimmed, get_color("settings_default"))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
if len(message_lines) > viewport_h:
|
||||
old_index = ui_state.start_index[4] if len(ui_state.start_index) > 4 else 0
|
||||
while len(ui_state.start_index) <= 4:
|
||||
ui_state.start_index.append(0)
|
||||
ui_state.start_index[4] = scroll_offset
|
||||
draw_main_arrows(msg_win, len(message_lines) - 1, window=4)
|
||||
ui_state.start_index[4] = old_index
|
||||
|
||||
try:
|
||||
ok_text = " Up/Down: Nodes PgUp/PgDn: Scroll Esc: Close "
|
||||
dialog_win.addstr(
|
||||
dialog_height - 2,
|
||||
max(1, (dialog_width - len(ok_text)) // 2),
|
||||
ok_text[: max(0, dialog_width - 2)],
|
||||
get_color("settings_default", reverse=True),
|
||||
)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
dialog_win.refresh()
|
||||
msg_win.noutrefresh()
|
||||
curses.doupdate()
|
||||
|
||||
dialog_win.timeout(200)
|
||||
char = dialog_win.getch()
|
||||
|
||||
if menu_state.need_redraw:
|
||||
menu_state.need_redraw = False
|
||||
continue
|
||||
|
||||
if char in (27, curses.KEY_LEFT, curses.KEY_ENTER, 10, 13, 32):
|
||||
break
|
||||
if char == curses.KEY_UP:
|
||||
old_selected_node = ui_state.selected_node
|
||||
ui_state.selected_node = (ui_state.selected_node - 1) % len(ui_state.node_list)
|
||||
scroll_offset = 0
|
||||
refresh_node_selection(old_selected_node, highlight=True)
|
||||
elif char == curses.KEY_DOWN:
|
||||
old_selected_node = ui_state.selected_node
|
||||
ui_state.selected_node = (ui_state.selected_node + 1) % len(ui_state.node_list)
|
||||
scroll_offset = 0
|
||||
refresh_node_selection(old_selected_node, highlight=True)
|
||||
elif char == curses.KEY_PPAGE:
|
||||
scroll_offset = max(0, scroll_offset - viewport_h)
|
||||
elif char == curses.KEY_NPAGE:
|
||||
scroll_offset = min(max_scroll, scroll_offset + viewport_h)
|
||||
elif char == curses.KEY_HOME:
|
||||
scroll_offset = 0
|
||||
elif char == curses.KEY_END:
|
||||
scroll_offset = max_scroll
|
||||
elif char == curses.KEY_RESIZE:
|
||||
continue
|
||||
|
||||
except KeyError:
|
||||
return
|
||||
finally:
|
||||
if dialog_win is not None:
|
||||
dialog_win.erase()
|
||||
dialog_win.refresh()
|
||||
ui_state.current_window = previous_window
|
||||
curses.curs_set(1)
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def handle_ctrl_t(stdscr: curses.window) -> None:
|
||||
@@ -624,6 +772,7 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
ui_state.current_window = 4
|
||||
settings_menu(stdscr, interface_state.interface)
|
||||
ui_state.current_window = previous_window
|
||||
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
|
||||
curses.curs_set(1)
|
||||
refresh_node_list()
|
||||
handle_resize(stdscr, False)
|
||||
@@ -828,7 +977,9 @@ def draw_channel_list() -> None:
|
||||
notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
|
||||
|
||||
# Truncate the channel name if it's too long to fit in the window
|
||||
truncated_channel = fit_text(f"{channel}{notification}", win_width - 3, suffix="-")
|
||||
truncated_channel = (
|
||||
(channel[: win_width - 5] + "-" if len(channel) > win_width - 5 else channel) + notification
|
||||
).ljust(win_width - 3)
|
||||
|
||||
color = get_color("channel_list")
|
||||
if idx == ui_state.selected_channel:
|
||||
@@ -863,7 +1014,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
|
||||
|
||||
row = 0
|
||||
for prefix, message in messages:
|
||||
full_message = f"{prefix}{message}"
|
||||
full_message = normalize_message_text(f"{prefix}{message}")
|
||||
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])
|
||||
@@ -905,10 +1056,8 @@ def draw_node_list() -> None:
|
||||
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)
|
||||
nodes_pad = curses.newpad(1, 1)
|
||||
if nodes_pad is None:
|
||||
nodes_pad = curses.newpad(1, 1)
|
||||
|
||||
try:
|
||||
nodes_pad.erase()
|
||||
@@ -922,27 +1071,12 @@ def draw_node_list() -> None:
|
||||
node = interface_state.interface.nodesByNum[node_num]
|
||||
secure = "user" in node and "publicKey" in node["user"] and node["user"]["publicKey"]
|
||||
status_icon = "🔐" if secure else "🔓"
|
||||
node_name = get_name_from_database(node_num, "long")
|
||||
user_name = node["user"]["shortName"]
|
||||
|
||||
uptime_str = ""
|
||||
if "deviceMetrics" in node and "uptimeSeconds" in node["deviceMetrics"]:
|
||||
uptime_str = f" / Up: {get_readable_duration(node['deviceMetrics']['uptimeSeconds'])}"
|
||||
|
||||
last_heard_str = f" ■ {get_time_ago(node['lastHeard'])}" if node.get("lastHeard") else ""
|
||||
hops_str = f" ■ Hops: {node['hopsAway']}" if "hopsAway" in node else ""
|
||||
snr_str = f" ■ SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
|
||||
node_name = get_node_display_name(node_num, node)
|
||||
|
||||
# Future node name custom formatting possible
|
||||
node_str = fit_text(f"{status_icon} {node_name}", box_width - 2)
|
||||
color = "node_list"
|
||||
if "isFavorite" in node and node["isFavorite"]:
|
||||
color = "node_favorite"
|
||||
if "isIgnored" in node and node["isIgnored"]:
|
||||
color = "node_ignored"
|
||||
nodes_pad.addstr(
|
||||
i, 1, node_str, get_color(color, reverse=ui_state.selected_node == i and ui_state.current_window == 2)
|
||||
)
|
||||
node_str = f"{status_icon} {node_name}"
|
||||
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
|
||||
nodes_pad.addstr(i, 1, node_str, get_node_row_color(i))
|
||||
|
||||
paint_frame(nodes_win, selected=(ui_state.current_window == 2))
|
||||
nodes_win.refresh()
|
||||
@@ -1063,16 +1197,9 @@ def draw_packetlog_win() -> None:
|
||||
span += column
|
||||
|
||||
# Add headers
|
||||
headers = " ".join(
|
||||
[
|
||||
fit_text("From", columns[0]),
|
||||
fit_text("To", columns[1]),
|
||||
fit_text("Port", columns[2]),
|
||||
fit_text("Payload", max(1, width - span - 3)),
|
||||
]
|
||||
)
|
||||
headers = f"{'From':<{columns[0]}} {'To':<{columns[1]}} {'Port':<{columns[2]}} {'Payload':<{width-span}}"
|
||||
packetlog_win.addstr(
|
||||
1, 1, fit_text(headers, width - 2), get_color("log_header", underline=True)
|
||||
1, 1, headers[: width - 2], get_color("log_header", underline=True)
|
||||
) # Truncate headers if they exceed window width
|
||||
|
||||
for i, packet in enumerate(reversed(ui_state.packet_buffer)):
|
||||
@@ -1080,22 +1207,22 @@ def draw_packetlog_win() -> None:
|
||||
break
|
||||
|
||||
# Format each field
|
||||
from_id = fit_text(get_name_from_database(packet["from"], "short"), columns[0])
|
||||
from_id = get_name_from_database(packet["from"], "short").ljust(columns[0])
|
||||
to_id = (
|
||||
fit_text("BROADCAST", columns[1])
|
||||
"BROADCAST".ljust(columns[1])
|
||||
if str(packet["to"]) == "4294967295"
|
||||
else fit_text(get_name_from_database(packet["to"], "short"), columns[1])
|
||||
else get_name_from_database(packet["to"], "short").ljust(columns[1])
|
||||
)
|
||||
if "decoded" in packet:
|
||||
port = fit_text(str(packet["decoded"].get("portnum", "")), columns[2])
|
||||
port = str(packet["decoded"].get("portnum", "")).ljust(columns[2])
|
||||
parsed_payload = parse_protobuf(packet)
|
||||
else:
|
||||
port = fit_text("NO KEY", columns[2])
|
||||
port = "NO KEY".ljust(columns[2])
|
||||
parsed_payload = "NO KEY"
|
||||
|
||||
# Combine and truncate if necessary
|
||||
logString = f"{from_id} {to_id} {port} {parsed_payload}"
|
||||
logString = fit_text(logString, width - 3)
|
||||
logString = logString[: width - 3]
|
||||
|
||||
# Add to the window
|
||||
packetlog_win.addstr(i + 2, 1, logString, get_color("log"))
|
||||
|
||||
@@ -1,18 +1,13 @@
|
||||
import curses
|
||||
import re
|
||||
import unicodedata
|
||||
from typing import Any, Optional, List, Dict
|
||||
|
||||
from wcwidth import wcwidth, wcswidth
|
||||
from unicodedata import east_asian_width
|
||||
|
||||
from contact.ui.colors import get_color
|
||||
from contact.utilities.i18n import t
|
||||
from contact.utilities.control_utils import transform_menu_path
|
||||
from typing import Any, Optional, List, Dict
|
||||
from contact.utilities.singleton import interface_state, ui_state
|
||||
|
||||
ZWJ = "\u200d"
|
||||
KEYCAP = "\u20e3"
|
||||
|
||||
|
||||
def get_node_color(node_index: int, reverse: bool = False):
|
||||
node_num = ui_state.node_list[node_index]
|
||||
@@ -332,158 +327,12 @@ def get_wrapped_help_text(
|
||||
return wrapped_help
|
||||
|
||||
|
||||
def _is_regional_indicator(char: str) -> bool:
|
||||
codepoint = ord(char)
|
||||
return 0x1F1E6 <= codepoint <= 0x1F1FF
|
||||
|
||||
|
||||
def _is_variation_selector(char: str) -> bool:
|
||||
codepoint = ord(char)
|
||||
return 0xFE00 <= codepoint <= 0xFE0F or 0xE0100 <= codepoint <= 0xE01EF
|
||||
|
||||
|
||||
def _is_emoji_modifier(char: str) -> bool:
|
||||
codepoint = ord(char)
|
||||
return 0x1F3FB <= codepoint <= 0x1F3FF
|
||||
|
||||
|
||||
def _is_display_modifier(char: str) -> bool:
|
||||
return unicodedata.category(char) in {"Mn", "Mc", "Me"} or _is_emoji_modifier(char)
|
||||
|
||||
|
||||
def iter_display_units(text: str) -> List[str]:
|
||||
"""Split text into display units so emoji sequences stay intact."""
|
||||
units: List[str] = []
|
||||
index = 0
|
||||
|
||||
while index < len(text):
|
||||
unit = text[index]
|
||||
index += 1
|
||||
|
||||
if _is_regional_indicator(unit) and index < len(text) and _is_regional_indicator(text[index]):
|
||||
unit += text[index]
|
||||
index += 1
|
||||
|
||||
while index < len(text) and _is_display_modifier(text[index]):
|
||||
unit += text[index]
|
||||
index += 1
|
||||
|
||||
while index < len(text) and text[index] == ZWJ and index + 1 < len(text):
|
||||
unit += text[index]
|
||||
index += 1
|
||||
unit += text[index]
|
||||
index += 1
|
||||
|
||||
while index < len(text) and _is_display_modifier(text[index]):
|
||||
unit += text[index]
|
||||
index += 1
|
||||
|
||||
units.append(unit)
|
||||
|
||||
return units
|
||||
|
||||
|
||||
def sanitize_for_curses(text: str) -> str:
|
||||
"""Collapse complex emoji sequences to stable fallbacks before rendering."""
|
||||
sanitized: List[str] = []
|
||||
|
||||
for unit in iter_display_units(text):
|
||||
if ZWJ not in unit and KEYCAP not in unit and not any(
|
||||
_is_variation_selector(char) or _is_emoji_modifier(char) for char in unit
|
||||
):
|
||||
sanitized.append(unit)
|
||||
continue
|
||||
|
||||
visible = [
|
||||
char
|
||||
for char in unit
|
||||
if char != ZWJ and char != KEYCAP and not _is_variation_selector(char) and not _is_emoji_modifier(char)
|
||||
]
|
||||
|
||||
if KEYCAP in unit and visible:
|
||||
sanitized.append(visible[0])
|
||||
elif ZWJ in unit and visible:
|
||||
sanitized.append(visible[0])
|
||||
elif any(_is_emoji_modifier(char) for char in unit) and visible:
|
||||
sanitized.append(visible[0])
|
||||
elif visible:
|
||||
sanitized.append("".join(visible))
|
||||
else:
|
||||
sanitized.append(unit)
|
||||
|
||||
return "".join(sanitized)
|
||||
|
||||
|
||||
def text_width(text: str) -> int:
|
||||
text = sanitize_for_curses(text)
|
||||
width = wcswidth(text)
|
||||
if width >= 0:
|
||||
return width
|
||||
return sum(max(wcwidth(char), 0) for char in text)
|
||||
|
||||
|
||||
def slice_text_to_width(text: str, max_width: int) -> str:
|
||||
"""Return the longest prefix that fits within max_width terminal cells."""
|
||||
text = sanitize_for_curses(text)
|
||||
if max_width <= 0:
|
||||
return ""
|
||||
|
||||
chunk = ""
|
||||
for unit in iter_display_units(text):
|
||||
candidate = chunk + unit
|
||||
if text_width(candidate) > max_width:
|
||||
break
|
||||
chunk = candidate
|
||||
|
||||
return chunk
|
||||
|
||||
|
||||
def fit_text(text: str, width: int, suffix: str = "") -> str:
|
||||
"""Trim and pad text so its terminal display width fits exactly."""
|
||||
text = sanitize_for_curses(text)
|
||||
suffix = sanitize_for_curses(suffix)
|
||||
if width <= 0:
|
||||
return ""
|
||||
|
||||
if text_width(text) > width:
|
||||
suffix = slice_text_to_width(suffix, width)
|
||||
available = max(0, width - text_width(suffix))
|
||||
text = slice_text_to_width(text, available).rstrip() + suffix
|
||||
|
||||
padding = max(0, width - text_width(text))
|
||||
return text + (" " * padding)
|
||||
|
||||
|
||||
def split_text_to_width(text: str, max_width: int) -> List[str]:
|
||||
"""Split text into chunks that each fit within max_width terminal cells."""
|
||||
text = sanitize_for_curses(text)
|
||||
if max_width <= 0:
|
||||
return [""]
|
||||
|
||||
chunks: List[str] = []
|
||||
chunk = ""
|
||||
|
||||
for unit in iter_display_units(text):
|
||||
candidate = chunk + unit
|
||||
if chunk and text_width(candidate) > max_width:
|
||||
chunks.append(chunk)
|
||||
chunk = unit
|
||||
else:
|
||||
chunk = candidate
|
||||
|
||||
if text_width(chunk) > max_width:
|
||||
chunks.append(slice_text_to_width(chunk, max_width))
|
||||
chunk = ""
|
||||
|
||||
if chunk:
|
||||
chunks.append(chunk)
|
||||
|
||||
return chunks or [""]
|
||||
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."""
|
||||
text = sanitize_for_curses(text)
|
||||
|
||||
whitespace = "\t\n\x0b\x0c\r "
|
||||
whitespace_trans = dict.fromkeys(map(ord, whitespace), ord(" "))
|
||||
@@ -497,18 +346,24 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
|
||||
wrap_width -= margin
|
||||
|
||||
for word in words:
|
||||
word_chunks = split_text_to_width(word, wrap_width) if text_width(word) > wrap_width else [word]
|
||||
word_length = text_width(word)
|
||||
|
||||
for chunk in word_chunks:
|
||||
chunk_length = text_width(chunk)
|
||||
|
||||
if line_length + chunk_length > wrap_width and chunk.strip():
|
||||
if word_length > wrap_width: # Break long words
|
||||
if line_buffer:
|
||||
wrapped_lines.append(line_buffer.strip())
|
||||
line_buffer = ""
|
||||
line_length = 0
|
||||
for i in range(0, word_length, wrap_width):
|
||||
wrapped_lines.append(word[i : i + wrap_width])
|
||||
continue
|
||||
|
||||
line_buffer += chunk
|
||||
line_length += chunk_length
|
||||
if line_length + word_length > wrap_width and word.strip():
|
||||
wrapped_lines.append(line_buffer.strip())
|
||||
line_buffer = ""
|
||||
line_length = 0
|
||||
|
||||
line_buffer += word
|
||||
line_length += word_length
|
||||
|
||||
if line_buffer:
|
||||
wrapped_lines.append(line_buffer.strip())
|
||||
|
||||
@@ -566,7 +566,7 @@ def save_json(file_path: str, data: Dict[str, Any]) -> None:
|
||||
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)
|
||||
config.reload_config()
|
||||
reload_translations(data.get("language"))
|
||||
|
||||
|
||||
|
||||
54
contact/utilities/emoji_utils.py
Normal file
54
contact/utilities/emoji_utils.py
Normal file
@@ -0,0 +1,54 @@
|
||||
"""Helpers for normalizing emoji sequences in width-sensitive message rendering."""
|
||||
|
||||
# Strip zero-width and presentation modifiers that make terminal cell width inconsistent.
|
||||
EMOJI_MODIFIER_REPLACEMENTS = {
|
||||
"\u200d": "",
|
||||
"\u20e3": "",
|
||||
"\ufe0e": "",
|
||||
"\ufe0f": "",
|
||||
"\U0001F3FB": "",
|
||||
"\U0001F3FC": "",
|
||||
"\U0001F3FD": "",
|
||||
"\U0001F3FE": "",
|
||||
"\U0001F3FF": "",
|
||||
}
|
||||
|
||||
_EMOJI_MODIFIER_TRANSLATION = str.maketrans(EMOJI_MODIFIER_REPLACEMENTS)
|
||||
_REGIONAL_INDICATOR_START = ord("\U0001F1E6")
|
||||
_REGIONAL_INDICATOR_END = ord("\U0001F1FF")
|
||||
|
||||
|
||||
def _regional_indicator_to_letter(char: str) -> str:
|
||||
return chr(ord("A") + ord(char) - _REGIONAL_INDICATOR_START)
|
||||
|
||||
|
||||
def _normalize_flag_emoji(text: str) -> str:
|
||||
"""Convert flag emoji built from regional indicators into ASCII country codes."""
|
||||
normalized = []
|
||||
index = 0
|
||||
|
||||
while index < len(text):
|
||||
current = text[index]
|
||||
current_ord = ord(current)
|
||||
|
||||
if _REGIONAL_INDICATOR_START <= current_ord <= _REGIONAL_INDICATOR_END and index + 1 < len(text):
|
||||
next_char = text[index + 1]
|
||||
next_ord = ord(next_char)
|
||||
if _REGIONAL_INDICATOR_START <= next_ord <= _REGIONAL_INDICATOR_END:
|
||||
normalized.append(_regional_indicator_to_letter(current))
|
||||
normalized.append(_regional_indicator_to_letter(next_char))
|
||||
index += 2
|
||||
continue
|
||||
|
||||
normalized.append(current)
|
||||
index += 1
|
||||
|
||||
return "".join(normalized)
|
||||
|
||||
|
||||
def normalize_message_text(text: str) -> str:
|
||||
"""Strip modifiers and rewrite flag emoji into stable terminal-friendly text."""
|
||||
if not text:
|
||||
return text
|
||||
|
||||
return _normalize_flag_emoji(text.translate(_EMOJI_MODIFIER_TRANSLATION))
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.4.20"
|
||||
version = "1.4.22"
|
||||
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"}
|
||||
@@ -9,8 +9,7 @@ license = "GPL-3.0-only"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9,<3.15"
|
||||
dependencies = [
|
||||
"meshtastic (>=2.7.5,<3.0.0)",
|
||||
"wcwidth (>=0.2.13,<1.0.0)"
|
||||
"meshtastic (>=2.7.5,<3.0.0)"
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
|
||||
@@ -1,2 +1 @@
|
||||
meshtastic
|
||||
wcwidth>=0.2.13,<1.0.0
|
||||
|
||||
Reference in New Issue
Block a user