Compare commits

..

1 Commits
UI ... 1.4.14

Author SHA1 Message Date
pdxlocations
ba713366bc bump version 2026-02-11 21:58:20 -08:00
8 changed files with 83 additions and 183 deletions

View File

@@ -5,10 +5,9 @@ import shutil
import time
import subprocess
import threading
from typing import Any, Dict, Optional
# Debounce notification sounds so a burst of queued messages only plays once.
_SOUND_DEBOUNCE_SECONDS = 0.8
_sound_timer: Optional[threading.Timer] = None
_sound_timer: threading.Timer | None = None
_sound_timer_lock = threading.Lock()
_last_sound_request = 0.0
@@ -43,6 +42,8 @@ def schedule_notification_sound(delay: float = _SOUND_DEBOUNCE_SECONDS) -> None:
_sound_timer = threading.Timer(delay, _fire, args=(now,))
_sound_timer.daemon = True
_sound_timer.start()
from typing import Any, Dict
from contact.utilities.utils import (
refresh_node_list,
add_new_message,

View File

@@ -12,7 +12,6 @@ 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 move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
@@ -20,9 +19,7 @@ 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).
@@ -65,48 +62,6 @@ 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) -> 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"
return get_color(color, reverse=index == ui_state.selected_node and ui_state.current_window == 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
@@ -206,24 +161,6 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
pass
def drain_resize_events(input_win: curses.window) -> Union[str, int, None]:
"""Wait for resize events to settle and preserve one queued non-resize key."""
input_win.timeout(RESIZE_DEBOUNCE_MS)
try:
while True:
try:
next_char = input_win.get_wch()
except curses.error:
return None
if next_char == curses.KEY_RESIZE:
continue
return next_char
finally:
input_win.timeout(-1)
def main_ui(stdscr: curses.window) -> None:
"""Main UI loop for the curses interface."""
global input_text
@@ -231,7 +168,6 @@ def main_ui(stdscr: curses.window) -> None:
root_win = stdscr
input_text = ""
queued_char = None
stdscr.keypad(True)
get_channels()
handle_resize(stdscr, True)
@@ -240,11 +176,7 @@ def main_ui(stdscr: curses.window) -> None:
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
# Get user input from entry window
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
char = entry_win.get_wch()
# draw_debug(f"Keypress: {char}")
@@ -292,9 +224,7 @@ def main_ui(stdscr: curses.window) -> None:
elif char == curses.KEY_RESIZE:
input_text = ""
queued_char = drain_resize_events(entry_win)
handle_resize(stdscr, False)
continue
elif char == chr(4): # Ctrl + D to delete current channel or node
handle_ctrl_d()
@@ -403,16 +333,31 @@ 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
if ui_state.single_pane_mode:
handle_resize(root_win, False)
return
handle_resize(root_win, False)
refresh_main_window(old_window, selected=False)
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)
if not ui_state.single_pane_mode:
draw_window_arrows(old_window)
refresh_main_window(ui_state.current_window, selected=True)
if ui_state.current_window == 0:
paint_frame(channel_win, selected=True)
refresh_pad(0)
elif ui_state.current_window == 1:
paint_frame(messages_win, selected=True)
refresh_pad(1)
elif ui_state.current_window == 2:
paint_frame(nodes_win, selected=True)
refresh_pad(2)
draw_window_arrows(ui_state.current_window)
@@ -433,16 +378,31 @@ def handle_function_keys(char: int) -> None:
return
ui_state.current_window = target
if ui_state.single_pane_mode:
handle_resize(root_win, False)
return
handle_resize(root_win, False)
refresh_main_window(old_window, selected=False)
if old_window == 0:
paint_frame(channel_win, selected=False)
refresh_pad(0)
elif old_window == 1:
paint_frame(messages_win, selected=False)
refresh_pad(1)
elif old_window == 2:
paint_frame(nodes_win, selected=False)
refresh_pad(2)
if not ui_state.single_pane_mode:
draw_window_arrows(old_window)
refresh_main_window(ui_state.current_window, selected=True)
if ui_state.current_window == 0:
paint_frame(channel_win, selected=True)
refresh_pad(0)
elif ui_state.current_window == 1:
paint_frame(messages_win, selected=True)
refresh_pad(1)
elif ui_state.current_window == 2:
paint_frame(nodes_win, selected=True)
refresh_pad(2)
draw_window_arrows(ui_state.current_window)
@@ -879,7 +839,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
row = 0
for prefix, message in messages:
full_message = normalize_message_text(f"{prefix}{message}")
full_message = 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])
@@ -921,8 +881,10 @@ def draw_node_list() -> None:
if ui_state.current_window != 2 and ui_state.single_pane_mode:
return
if nodes_pad is None:
nodes_pad = curses.newpad(1, 1)
# 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)
try:
nodes_pad.erase()
@@ -936,12 +898,28 @@ 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_node_display_name(node_num, node)
node_name = get_name_from_database(node_num, "long")
user_name = node["user"]["shortName"]
uptime_str = ""
if "deviceMetrics" in node and "uptimeSeconds" in node["deviceMetrics"]:
uptime_str = f" / Up: {get_readable_duration(node['deviceMetrics']['uptimeSeconds'])}"
last_heard_str = f"{get_time_ago(node['lastHeard'])}" if node.get("lastHeard") else ""
hops_str = f" ■ Hops: {node['hopsAway']}" if "hopsAway" in node else ""
snr_str = f" ■ SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
# Future node name custom formatting possible
node_str = f"{status_icon} {node_name}"
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
nodes_pad.addstr(i, 1, node_str, get_node_row_color(i))
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)
)
paint_frame(nodes_win, selected=(ui_state.current_window == 2))
nodes_win.refresh()

View File

@@ -56,13 +56,6 @@ config_folder = os.path.abspath(config.node_configs_file_path)
# Load translations
field_mapping, help_text = parse_ini_file(translation_file)
def _is_repeated_field(field_desc) -> bool:
"""Return True if the protobuf field is repeated.
Protobuf 6.31.0 and later use an is_repeated property, while older versions compare against the label field.
"""
if hasattr(field_desc, "is_repeated"):
return bool(field_desc.is_repeated)
return field_desc.label == field_desc.LABEL_REPEATED
def reload_translations() -> None:
global translation_file, field_mapping, help_text
@@ -133,16 +126,12 @@ def display_menu() -> tuple[object, object]:
if full_key.startswith("config.network.ipv4_config.") and option in {"ip", "gateway", "subnet", "dns"}:
if isinstance(current_value, int):
try:
current_value = str(
ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False))
)
current_value = str(ipaddress.IPv4Address(current_value))
except ipaddress.AddressValueError:
pass
elif isinstance(current_value, str) and current_value.isdigit():
try:
current_value = str(
ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False))
)
current_value = str(ipaddress.IPv4Address(int(current_value)))
except ipaddress.AddressValueError:
pass
@@ -571,7 +560,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
new_value = new_value == "True" or new_value is True
menu_state.start_index.pop()
elif _is_repeated_field(field): # Handle repeated field - Not currently used
elif field.label == field.LABEL_REPEATED: # Handle repeated field - Not currently used
new_value = get_repeated_input(current_value)
new_value = current_value if new_value is None else new_value.split(", ")
menu_state.start_index.pop()

View File

@@ -9,17 +9,6 @@ from meshtastic.util import camel_to_snake, snake_to_camel, fromStr
# defs are from meshtastic/python/main
def _is_repeated_field(field_desc) -> bool:
"""Return True if the protobuf field is repeated.
Protobuf 6.31.0+ exposes `is_repeated`, while older versions require
checking `label == LABEL_REPEATED`.
"""
if hasattr(field_desc, "is_repeated"):
return bool(field_desc.is_repeated)
return field_desc.label == field_desc.LABEL_REPEATED
def traverseConfig(config_root, config, interface_config) -> bool:
"""Iterate through current config level preferences and either traverse deeper if preference is a dict or set preference"""
snake_name = camel_to_snake(config_root)
@@ -100,7 +89,7 @@ def setPref(config, comp_name, raw_val) -> bool:
return False
# repeating fields need to be handled with append, not setattr
if not _is_repeated_field(pref):
if pref.label != pref.LABEL_REPEATED:
try:
if config_type.message_type is not None:
config_values = getattr(config_part, config_type.name)

View File

@@ -1,54 +0,0 @@
"""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))

View File

@@ -462,10 +462,7 @@ from contact.utilities.singleton import menu_state # Ensure this is imported
def get_fixed32_input(current_value: int) -> int:
original_value = current_value
try:
ip_string = str(ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False)))
except Exception:
ip_string = str(ipaddress.IPv4Address(current_value))
ip_string = str(ipaddress.IPv4Address(current_value))
height = 10
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
@@ -527,7 +524,7 @@ def get_fixed32_input(current_value: int) -> int:
if len(octets) == 4 and all(octet.isdigit() and 0 <= int(octet) <= 255 for octet in octets):
curses.noecho()
curses.curs_set(0)
return int.from_bytes(ipaddress.IPv4Address(user_input).packed, "little", signed=False)
return int(ipaddress.ip_address(user_input))
else:
fixed32_win.addstr(
7,

View File

@@ -68,19 +68,19 @@ def get_chunks(data):
# Leave it string as last resort
value = value
# Python 3.9-compatible alternative to match/case.
if key == "uptime_seconds":
match key:
# convert seconds to hours, for our sanity
value = round(value / 60 / 60, 1)
elif key in ("longitude_i", "latitude_i"):
case "uptime_seconds":
value = round(value / 60 / 60, 1)
# Convert position to degrees (humanize), as per Meshtastic protobuf comment for this telemetry
# truncate to 6th digit after floating point, which would be still accurate
value = round(value * 1e-7, 6)
elif key == "wind_direction":
case "longitude_i" | "latitude_i":
value = round(value * 1e-7, 6)
# Convert wind direction from degrees to abbreviation
value = humanize_wind_direction(value)
elif key == "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
case "wind_direction":
value = humanize_wind_direction(value)
case "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
if key in sensors:
parsed+= f"{sensors[key.strip()]['icon']}{value}{sensors[key]['unit']} "

View File

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