Compare commits

..

1 Commits

Author SHA1 Message Date
pdxlocations
ba713366bc bump version 2026-02-11 21:58:20 -08:00
9 changed files with 51 additions and 160 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

@@ -14,12 +14,11 @@ from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
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
@@ -162,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
@@ -187,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)
@@ -196,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}")
@@ -248,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()
@@ -828,7 +802,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:
@@ -934,7 +910,8 @@ def draw_node_list() -> None:
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 = fit_text(f"{status_icon} {node_name}", box_width - 2)
node_str = f"{status_icon} {node_name}"
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
color = "node_list"
if "isFavorite" in node and node["isFavorite"]:
color = "node_favorite"
@@ -1063,16 +1040,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 +1050,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"))

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

@@ -1,12 +1,11 @@
import curses
import re
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
@@ -329,65 +328,7 @@ def get_wrapped_help_text(
def text_width(text: str) -> int:
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."""
if max_width <= 0:
return ""
chunk = ""
for char in text:
candidate = chunk + char
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."""
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."""
if max_width <= 0:
return [""]
chunks: List[str] = []
chunk = ""
for char in text:
candidate = chunk + char
if chunk and text_width(candidate) > max_width:
chunks.append(chunk)
chunk = char
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]:
@@ -405,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())

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

@@ -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.19"
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"}
@@ -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]

View File

@@ -1,2 +1 @@
meshtastic
wcwidth>=0.2.13,<1.0.0