Compare commits

...
Author SHA1 Message Date
pdxlocations afbde3c2d6 0 = Default 2026-09-09 22:55:35 -07:00
pdxlocations 51351d316f Merge pull request #310 from pdxlocations:ham
Add ham params
2026-09-09 22:50:22 -07:00
pdxlocations 888108f14e Add ham params 2026-09-09 22:50:02 -07:00
pdxlocations 72f97bfde7 Enhance telemetry beautification by adding local stats parsing and new metrics; update tests for coverage. 2026-09-06 00:02:25 -07:00
pdxlocations 31bff496f1 Bump version to 1.6.12 in pyproject.toml 2026-09-05 13:51:05 -07:00
pdxlocations b7c3ed20f2 retore missing commit 2026-09-05 13:49:56 -07:00
pdxlocations 7f7e0598d2 restore missing commits 2026-09-05 13:48:09 -07:00
pdxlocations 82e7456248 Bump version to 1.6.11 in pyproject.toml 2026-09-05 13:41:36 -07:00
pdxlocations 7b21c9c886 Enhance channel list rendering to display configured names without altering slot keys; add unit test for verification. 2026-09-05 13:41:19 -07:00
pdxlocations 392d236a40 Bump version to 1.6.10 in pyproject.toml 2026-09-05 13:36:37 -07:00
pdxlocations 546fab099c Merge pull request #308 from pdxlocations:flicker-fix
Refactor paint_frame and refresh_node_selection functions to include optional refresh parameter; add stage_node_details_background function for improved window management in F5 key handling tests.
2026-09-05 13:36:06 -07:00
pdxlocations a89d4f8c0d Refactor paint_frame and refresh_node_selection functions to include optional refresh parameter; add stage_node_details_background function for improved window management in F5 key handling tests. 2026-09-05 13:35:36 -07:00
pdxlocations 808fee4fe0 Bump version to 1.6.9 in pyproject.toml 2026-09-05 00:00:11 -07:00
pdxlocations 94d0a9053e Enhance interface initialization and splash screen status display
- Updated `initialize_runtime_interface` to accept `stdscr` for better integration with the UI.
- Modified `initialize_interface` to include a status callback for improved user feedback during connection attempts.
- Refactored `draw_splash` to dynamically display status messages, improving user experience during startup.
2026-09-04 23:59:53 -07:00
pdxlocations f38167db1a Bump version to 1.6.8 in pyproject.toml 2026-09-04 23:51:29 -07:00
pdxlocations 57caf463bc Merge pull request #305 from pdxlocations:edit-message-text
Edit message text with arrows
2026-09-04 23:30:23 -07:00
21 changed files with 612 additions and 69 deletions

No files matched your search

+6 -11
View File
@@ -38,6 +38,7 @@ from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.ui.dialog import dialog
from contact.utilities.interfaces import initialize_interface, reconnect_interface
from contact.utilities.logging_utils import configure_logging
from contact.utilities.utils import get_channels, get_nodeNum, get_node_list
from contact.utilities.singleton import ui_state, interface_state, app_state
@@ -51,15 +52,7 @@ os.environ.setdefault("TERM", "xterm-256color")
if os.environ.get("COLORTERM") == "gnome-terminal":
os.environ["TERM"] = "xterm-256color"
logging.basicConfig(
filename=config.log_file_path,
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
# Meshtastic or another dependency may have installed a stderr handler
# before Contact starts curses. Replace it so background warnings cannot
# overwrite the terminal UI.
force=True,
)
configure_logging(logging.INFO)
app_state.lock = threading.Lock()
@@ -121,7 +114,7 @@ def interface_is_ready(interface: object) -> bool:
def initialize_runtime_interface_with_retry(stdscr: curses.window, args: object):
while True:
interface = initialize_runtime_interface(args)
interface = initialize_runtime_interface(args, stdscr)
if getattr(args, "demo_screenshot", False) or interface_is_ready(interface):
return interface
@@ -162,10 +155,12 @@ def initialize_globals(seed_demo: bool = False) -> None:
load_messages_from_db()
def initialize_runtime_interface(args: object):
def initialize_runtime_interface(args: object, stdscr=None):
if getattr(args, "demo_screenshot", False):
configure_demo_database()
return build_demo_interface()
if stdscr is not None:
return initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
return initialize_interface(args)
+7 -1
View File
@@ -98,7 +98,12 @@ dialog.resize_body, "Please resize the terminal to at least {rows} rows.", ""
user, "User"
longName, "Node long name", "If you are a licensed HAM operator and have enabled HAM mode, this must be set to your HAM operator call sign."
shortName, "Node short name", "Must be up to 4 bytes. Usually this is 4 characters, if using latin characters and no emojis."
isLicensed, "Enable licensed amateur (HAM) mode", "IMPORTANT: Read Meshtastic help documentation before enabling."
isLicensed, "Enable licensed amateur (HAM) mode", "Shows HAM parameters for editing. Saving applies licensed operation, including radio and channel changes. Disabling does not restore previous settings."
call_sign, "Callsign", "Licensed operator callsign."
tx_power, "TX power in dBm", "Transmit power at the LoRa transceiver, before amplification. Use a value supported by your radio."
frequency, "Frequency in MHz", "Operating frequency override. Zero uses automatic frequency selection."
short_name, "HAM short name", "Optional short display name."
long_name, "Name appended to callsign", "Optional name appended to the callsign. Requires supporting firmware."
isUnmessagable, "Disable incoming messages", "When enabled, this node advertises that it cannot receive messages."
[app_settings]
@@ -160,6 +165,7 @@ node_ignored, "Node ignored", ""
[Channels.channel]
title, "Channels"
enabled, "Enabled", "Enable or disable this channel without clearing its name or encryption key."
channel_num, "Channel number", "The index number of this channel."
psk, "PSK", "The channel's encryption key."
name, "Name", "The channel's name."
+3 -7
View File
@@ -14,6 +14,7 @@ from contact.utilities.arg_parser import setup_parser
from contact.utilities.i18n import t
from contact.utilities.input_handlers import get_list_input
from contact.utilities.interfaces import initialize_interface, reconnect_interface
from contact.utilities.logging_utils import configure_logging
def close_interface(interface: object) -> None:
@@ -36,7 +37,7 @@ def main(stdscr: curses.window) -> None:
parser = setup_parser()
args = parser.parse_args()
interface = initialize_interface(args)
interface = initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
if interface.localNode.localConfig.lora.region == 0:
confirmation = get_list_input(
@@ -81,12 +82,7 @@ def ensure_min_rows(stdscr: curses.window, min_rows: int = 11) -> None:
stdscr.refresh()
logging.basicConfig( # Run `tail -f client.log` in another terminal to view live
filename=config.log_file_path,
level=logging.WARNING, # DEBUG, INFO, WARNING, ERROR, CRITICAL)
format="%(asctime)s - %(levelname)s - %(message)s",
force=True,
)
configure_logging(logging.WARNING)
if __name__ == "__main__":
log_file = config.log_file_path
+54 -18
View File
@@ -246,13 +246,14 @@ def compute_widths(total_w: int, focus: int):
return MIN_COL, MIN_COL, total_w - 2 * MIN_COL
def paint_frame(win, selected: bool) -> None:
def paint_frame(win, selected: bool, refresh: bool = True) -> None:
if win is not entry_win and ui_state.input_focused:
selected = False
win.attrset(get_color("window_frame_selected") if selected else get_color("window_frame"))
win.box()
win.attrset(get_color("window_frame"))
win.refresh()
if refresh:
win.refresh()
def get_channel_row_color(index: int) -> int:
@@ -275,7 +276,7 @@ def get_node_row_color(index: int, highlight: bool = False) -> int:
return get_color(color, reverse=reverse)
def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None:
def refresh_node_selection(old_index: int = -1, highlight: bool = False, refresh: bool = True) -> None:
if nodes_pad is None or not ui_state.node_list:
return
@@ -294,8 +295,9 @@ def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None
pass
ui_state.start_index[2] = max(0, ui_state.selected_node - (nodes_win.getmaxyx()[0] - 3))
refresh_pad(2)
draw_window_arrows(2)
if refresh:
refresh_pad(2)
draw_window_arrows(2)
def refresh_main_window(window_id: int, selected: bool) -> None:
@@ -306,7 +308,7 @@ def refresh_main_window(window_id: int, selected: bool) -> None:
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)
paint_frame(messages_win, selected=selected, refresh=False)
refresh_pad(1)
elif window_id == 2:
paint_frame(nodes_win, selected=selected)
@@ -700,7 +702,7 @@ def draw_message_input(draft: str) -> None:
used += text_width(content[start])
visible = slice_to_width(content[start:], available)
draw_text_field(entry_win, prompt + visible, get_color("input"))
paint_frame(entry_win, selected=ui_state.input_focused)
paint_frame(entry_win, selected=ui_state.input_focused, refresh=False)
try:
hint = " Ctrl+K Help "
if width >= len(hint) + 4:
@@ -935,6 +937,28 @@ def handle_enter(input_text: str) -> str:
return input_text
def stage_node_details_background(stdscr, previous_window):
"""Restore the panes behind a moved dialog without displaying an intermediate frame."""
current_window = ui_state.current_window
try:
ui_state.current_window = previous_window
stdscr.touchwin()
stdscr.noutrefresh()
windows = (channel_win, messages_win, nodes_win)
for index, win in enumerate(windows):
if ui_state.single_pane_mode and index != previous_window:
continue
win.touchwin()
refresh_pad(index, commit=False)
if ui_state.display_log and (not ui_state.single_pane_mode or previous_window == 1):
packetlog_win.touchwin()
packetlog_win.noutrefresh()
entry_win.touchwin()
entry_win.noutrefresh()
finally:
ui_state.current_window = current_window
def handle_f5_key(stdscr: curses.window) -> None:
if not ui_state.node_list:
return
@@ -947,7 +971,10 @@ def handle_f5_key(stdscr: curses.window) -> None:
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')}")
message_parts.append(f"• Role: {node.get('user', {}).get('role', 'Unknown')}")
user = node.get('user', {})
# Protobuf dictionaries omit CLIENT because it is the default enum value.
role = user.get('role', 'CLIENT') if user else 'Unknown'
message_parts.append(f"• Role: {role}")
message_parts.append(f"Public key: {node.get('user', {}).get('publicKey')}")
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
@@ -1032,6 +1059,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
ui_state.current_window = 4
scroll_offset = 0
dialog_win = None
previous_layout = None
curses.curs_set(0)
refresh_node_selection(highlight=True)
@@ -1051,11 +1079,15 @@ def handle_f5_key(stdscr: curses.window) -> None:
max_scroll = max(0, len(message_lines) - viewport_h)
scroll_offset = max(0, min(scroll_offset, max_scroll))
layout = (dialog_height, dialog_width, y, x, ui_state.selected_node, height, width)
if previous_layout is not None and layout != previous_layout:
stage_node_details_background(stdscr, previous_window)
previous_layout = layout
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)
@@ -1100,7 +1132,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
except curses.error:
pass
dialog_win.refresh()
dialog_win.noutrefresh()
msg_win.noutrefresh()
curses.doupdate()
@@ -1120,12 +1152,12 @@ def handle_f5_key(stdscr: curses.window) -> None:
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)
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
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)
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
elif char == curses.KEY_PPAGE:
scroll_offset = max(0, scroll_offset - viewport_h)
elif char == curses.KEY_NPAGE:
@@ -1642,7 +1674,7 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
messages_pad.addstr(row, 1, line, color)
ui_state.message_line_ranges[channel] = message_ranges
paint_frame(messages_win, selected=(ui_state.current_window == 1))
paint_frame(messages_win, selected=(ui_state.current_window == 1), refresh=False)
visible_lines = get_msg_window_lines(messages_win, packetlog_win)
@@ -1657,7 +1689,6 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
refresh_message_highlight()
messages_win.refresh()
refresh_pad(1)
draw_packetlog_win()
draw_window_arrows(1)
@@ -1987,7 +2018,7 @@ def search(win: int) -> None:
entry_win.erase()
def refresh_pad(window: int) -> None:
def refresh_pad(window: int, commit: bool = True) -> None:
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
if ui_state.single_pane_mode and window != ui_state.current_window:
@@ -2004,7 +2035,7 @@ def refresh_pad(window: int) -> None:
if ui_state.display_log:
packetlog_win.box()
packetlog_win.refresh()
packetlog_win.noutrefresh()
elif window == 2:
pad = nodes_pad
@@ -2047,9 +2078,12 @@ def refresh_pad(window: int) -> None:
return
draw_frame_title(box, get_window_title(window))
box.refresh()
box.noutrefresh()
pad.refresh(
# The frame overlaps the pad's viewport. Recompose even unchanged pad
# rows before committing, so its text cannot flash blank between draws.
pad.touchwin()
pad.noutrefresh(
start_index,
0,
top,
@@ -2057,6 +2091,8 @@ def refresh_pad(window: int) -> None:
bottom,
right,
)
if commit:
curses.doupdate()
def add_notification(channel_number: int) -> None:
+37 -5
View File
@@ -25,7 +25,7 @@ from contact.utilities.input_handlers import (
)
from contact.ui.colors import get_color
from contact.ui.dialog import dialog
from contact.ui.menus import generate_menu_from_protobuf
from contact.ui.menus import generate_menu_from_protobuf, update_ham_fields
from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
from contact.ui.user_config import json_editor
from contact.utilities.arg_parser import setup_parser
@@ -141,6 +141,12 @@ def display_menu() -> tuple[object, object]:
current_value = field_info[1] if isinstance(field_info, tuple) else ""
full_key = ".".join(transformed_path + [option])
display_name = field_mapping.get(full_key, option)
if menu_state.menu_path == ["Main Menu", "Channels"] and isinstance(field_info, dict):
# Keep slot keys for navigation/saving, but label rows with the
# configured name. Empty names retain their slot label.
channel_name = field_info.get("name", (None, ""))[1]
if isinstance(channel_name, str) and channel_name.strip():
display_name = channel_name
if full_key.startswith("config.network.ipv4_config.") and option in {"ip", "gateway", "subnet", "dns"}:
if isinstance(current_value, int):
@@ -354,6 +360,14 @@ def _request_remote_section(
_request_remote_channels_with_timeout(node, cancel_callback=cancel_callback)
return True
if menu_path == ["Main Menu"] and selected_option == "User Settings":
# HAM inputs derive their initial power/frequency from this node's LoRa config.
field = node.localConfig.DESCRIPTOR.fields_by_name["lora"]
if status_callback:
status_callback("Requesting lora config…")
_request_remote_with_timeout(node.requestConfig, field, cancel_callback=cancel_callback)
return True
if len(menu_path) != 2:
return False
@@ -502,6 +516,7 @@ def settings_menu(
break
if reconnect_required:
interface = reconnect_interface_with_splash(stdscr, interface)
node = interface.localNode
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
if len(menu_state.menu_path) > 1:
@@ -756,9 +771,6 @@ def settings_menu(
new_value = new_value == "True"
menu_state.current_menu[selected_option] = (field, new_value)
for option, (field, value) in menu_state.current_menu.items():
modified_settings[option] = value
menu_state.start_index.pop()
elif selected_option in ["latitude", "longitude", "altitude"]:
@@ -779,6 +791,11 @@ def settings_menu(
new_value = current_value if new_values is None else [base64.b64decode(key) for key in new_values]
menu_state.start_index.pop()
elif field is None and selected_option == "enabled" and menu_state.menu_path[1] == "Channels":
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
new_value = new_value == "True" or new_value is True
menu_state.start_index.pop()
elif field.type == 8: # Handle boolean type
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
if new_value == "Not Set":
@@ -820,8 +837,17 @@ def settings_menu(
else: # Handle other field types
input_type = get_input_type_for_field(field)
allow_empty = (
menu_state.menu_path == ["Main Menu", "User Settings"]
and selected_option in ("short_name", "long_name")
) or (
selected_option == "name"
and len(menu_state.menu_path) == 3
and menu_state.menu_path[1] == "Channels"
)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
f"{human_readable_name} is currently: {current_value}", selected_option, input_type,
allow_empty=allow_empty,
)
new_value = current_value if new_value is None else new_value
menu_state.start_index.pop()
@@ -846,6 +872,12 @@ def settings_menu(
new_value = enum_value_descriptor.name if enum_value_descriptor else new_value
menu_state.current_menu[selected_option] = (field, new_value)
if menu_state.menu_path == ["Main Menu", "User Settings"]:
update_ham_fields(menu_state.current_menu, node)
# Save the full HAM command, including fields that were not edited.
modified_settings.clear()
modified_settings.update({name: setting[1] for name, setting in menu_state.current_menu.items()})
menu_state.selected_index = list(menu_state.current_menu).index(selected_option)
else:
requested_remote_section = False
if remote:
+38 -1
View File
@@ -5,7 +5,7 @@ from collections import OrderedDict
from typing import Any, Union, Dict
from google.protobuf.message import Message
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
from meshtastic.protobuf import admin_pb2, channel_pb2, config_pb2, module_config_pb2
def encode_if_bytes(value: Any) -> str:
"""Encode byte values to base64 string."""
@@ -57,6 +57,38 @@ def extract_fields(
return menu
def update_ham_fields(user_menu, node):
"""Reflect the selected license flag using the installed HAM descriptors."""
params = admin_pb2.HamParameters()
fields = params.DESCRIPTOR.fields_by_name
licensed = user_menu.get("isLicensed", (None, False))[1] in (True, "True")
if not licensed:
call_sign = user_menu.get("call_sign", (None, ""))[1]
suffix = user_menu.get("long_name", (None, ""))[1]
user_menu.setdefault("longName", (None, " ".join(filter(None, (call_sign, suffix)))))
user_menu.setdefault("shortName", (None, user_menu.get("short_name", (None, ""))[1]))
for name in fields:
user_menu.pop(name, None)
return
owner_name = user_menu.get("longName", (None, ""))[1]
call_sign, _, suffix = owner_name.partition(" ")
defaults = {
"call_sign": call_sign,
"short_name": user_menu.get("shortName", (None, ""))[1],
"long_name": suffix,
"tx_power": node.localConfig.lora.tx_power,
"frequency": node.localConfig.lora.override_frequency,
}
for field in params.DESCRIPTOR.fields:
if field.name in defaults:
setattr(params, field.name, defaults[field.name])
for name, setting in extract_fields(params, params).items():
user_menu.setdefault(name, setting)
user_menu.pop("longName", None)
user_menu.pop("shortName", None)
def generate_menu_from_protobuf(interface: object, node: Any = None, include_app_settings: bool = True) -> Dict[str, Any]:
"""
Builds the full settings menu structure from the protobuf definitions.
@@ -78,6 +110,7 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
"isLicensed": (None, current_user_config.get("isLicensed", "False")),
"isUnmessagable": (None, current_user_config.get("isUnmessagable", "False")),
}
update_ham_fields(menu_structure["Main Menu"]["User Settings"], node)
else:
logging.info("User settings not found in Node Info")
menu_structure["Main Menu"]["User Settings"] = "No user settings available"
@@ -93,6 +126,10 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
current_channel = node.getChannelByChannelIndex(i)
if current_channel:
channel_config = extract_fields(channel, current_channel.settings)
channel_config = {
"enabled": (None, current_channel.role != channel_pb2.Channel.Role.DISABLED),
**channel_config,
}
menu_structure["Main Menu"]["Channels"][f"Channel {i + 1}"] = channel_config
# Add Radio Settings
+10 -5
View File
@@ -1,8 +1,9 @@
import curses
import textwrap
from contact.ui.colors import get_color
def draw_splash(stdscr: object) -> None:
def draw_splash(stdscr: object, status: str = "connecting...") -> None:
"""Draw the splash screen with a logo and connecting message."""
curses.curs_set(0)
@@ -13,16 +14,20 @@ def draw_splash(stdscr: object) -> None:
message_1 = "/ Λ"
message_2 = "/ / \\"
message_3 = "P W R D"
message_4 = "connecting..."
start_x = width // 2 - len(message_1) // 2
start_x2 = width // 2 - len(message_4) // 2
start_y = height // 2 - 1
stdscr.addstr(start_y, start_x, message_1, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 1, start_x - 1, message_2, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 2, start_x - 2, message_3, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 4, start_x2, message_4, get_color("splash_text"))
stdscr.move(start_y + 5, start_x2)
status_lines = []
for line in status.splitlines():
status_lines.extend(textwrap.wrap(line, max(1, width - 4)) or [""])
for offset, line in enumerate(status_lines):
row = start_y + 4 + offset
if row >= height - 1:
break
stdscr.addstr(row, max(1, (width - len(line)) // 2), line, get_color("splash_text"))
stdscr.attrset(get_color("window_frame"))
stdscr.box()
+4 -1
View File
@@ -38,7 +38,7 @@ def invalid_input(window: curses.window, message: str, redraw_func: Optional[cal
curses.curs_set(1)
def get_text_input(prompt: str, selected_config: str, input_type: str) -> Optional[str]:
def get_text_input(prompt: str, selected_config: str, input_type: str, *, allow_empty: bool = False) -> Optional[str]:
"""Handles user input with wrapped text for long prompts."""
def redraw_input_win():
@@ -127,6 +127,9 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
menu_state.need_redraw = True
if not user_input.strip():
if allow_empty:
user_input = ""
break
invalid_input(
input_win,
t("ui.error.value_empty", default="Value cannot be empty."),
+9 -2
View File
@@ -1,4 +1,6 @@
import logging
import contextlib
import io
import time
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
@@ -25,7 +27,7 @@ def interface_is_connected(interface) -> bool:
return getattr(interface, "localNode", None) is not None
def initialize_interface(args):
def initialize_interface(args, status_callback=None):
try:
if args.ble:
@@ -43,7 +45,10 @@ def initialize_interface(args):
logging.error(f"Error connecting to {args.host}. {ex}")
else:
try:
client = meshtastic.serial_interface.SerialInterface(args.port)
# The library prints its fallback notice directly to stdout,
# which would overwrite the curses splash screen.
with contextlib.redirect_stdout(io.StringIO()):
client = meshtastic.serial_interface.SerialInterface(args.port)
except FileNotFoundError as ex:
logging.error(f"The serial device at '{args.port}' was not found. {ex}")
except PermissionError as ex:
@@ -56,6 +61,8 @@ def initialize_interface(args):
logging.error(f"The serial device couldn't be opened, it might be in use by another process. {ex}")
if client.devPath is None:
try:
if status_callback:
status_callback("No serial device found.\nConnecting via TCP to localhost…")
client = meshtastic.tcp_interface.TCPInterface("localhost")
except Exception as ex:
logging.error(f"Error connecting to localhost:{ex}")
+26
View File
@@ -0,0 +1,26 @@
import logging
import os
import sys
import contact.ui.default_config as config
def configure_logging(level: int) -> None:
"""Set up file logging or exit with instructions for fixing its path."""
try:
logging.basicConfig(
filename=config.log_file_path,
level=level,
format="%(asctime)s - %(levelname)s - %(message)s",
force=True,
)
except OSError as error:
print(
f"Unable to open log file: {os.path.abspath(config.log_file_path)}\n"
f"{error}\n"
f"Please edit config.json at: {os.path.abspath(config.json_file_path)}\n"
'Set "log_file_path" to a file in an existing, writable directory, '
"then restart Contact.",
file=sys.stderr,
)
raise SystemExit(1) from None
+25 -7
View File
@@ -1,4 +1,4 @@
from meshtastic.protobuf import channel_pb2
from meshtastic.protobuf import admin_pb2, channel_pb2
from google.protobuf.message import Message
import logging
import base64
@@ -139,6 +139,21 @@ def save_changes(interface, modified_settings, menu_state, node=None):
if is_unmessagable is not None:
is_unmessagable = is_unmessagable == "True" or is_unmessagable is True
if is_licensed and "call_sign" in modified_settings:
message = admin_pb2.AdminMessage()
for field in message.set_ham_mode.DESCRIPTOR.fields:
if field.name in modified_settings:
setattr(message.set_ham_mode, field.name, modified_settings[field.name])
node.ensureSessionKey()
# Keep the separate user flag and HAM command in one settings transaction.
node.beginSettingsTransaction()
if is_unmessagable is not None:
node.setOwner(None, None, True, is_unmessagable)
callback = None if node is interface.localNode else node.onAckNak
node._sendAdmin(message, onResponse=callback)
node.commitSettingsTransaction()
return True
node.setOwner(long_name, short_name, is_licensed, is_unmessagable)
logging.info(
@@ -160,7 +175,15 @@ def save_changes(interface, modified_settings, menu_state, node=None):
channel = node.channels[channel_num]
for key, value in modified_settings.items():
if key == "psk": # Special case: decode Base64 for psk
if key == "enabled":
if not value:
channel.role = channel_pb2.Channel.Role.DISABLED
elif channel.role == channel_pb2.Channel.Role.DISABLED:
channel.role = (
channel_pb2.Channel.Role.PRIMARY if channel_num == 0
else channel_pb2.Channel.Role.SECONDARY
)
elif key == "psk": # Special case: decode Base64 for psk
channel.settings.psk = base64.b64decode(value)
elif key in {"position_precision", "is_muted"}:
# These fields belong to ChannelSettings.module_settings,
@@ -169,11 +192,6 @@ def save_changes(interface, modified_settings, menu_state, node=None):
else:
setattr(channel.settings, key, value) # Use setattr for other fields
if channel_num == 0:
channel.role = channel_pb2.Channel.Role.PRIMARY
else:
channel.role = channel_pb2.Channel.Role.SECONDARY
node.writeChannel(channel_num)
logging.info(f"Updated Channel {channel_num} in {config_category}")
+34 -8
View File
@@ -13,9 +13,29 @@ sensors = {
'channel_utilization': {'icon':'ChUtil:', 'unit':'%'},
'air_util_tx': {'icon':'AirUtil:', 'unit':'%'},
'uptime_seconds': {'icon':'🆙 ', 'unit':'h'},
'num_packets_tx': {'icon':'📤 Tx:', 'unit':''},
'num_packets_rx': {'icon':'📥 Rx:', 'unit':''},
'num_packets_rx_bad': {'icon':'⚠️ RxBad:', 'unit':''},
'num_online_nodes': {'icon':'📡 Online:', 'unit':''},
'num_total_nodes': {'icon':'📡 Total:', 'unit':''},
'num_rx_dupe': {'icon':'📥 Dupe:', 'unit':''},
'num_tx_relay': {'icon':'📤 Relay:', 'unit':''},
'num_tx_relay_canceled': {'icon':'📤 Canceled:', 'unit':''},
'num_tx_dropped': {'icon':'⚠️ Dropped:', 'unit':''},
'heap_total_bytes': {'icon':'💾 Total:', 'unit':'B'},
'heap_free_bytes': {'icon':'💾 Free:', 'unit':'B'},
'noise_floor': {'icon':'📶 Noise:', 'unit':'dBm'},
'latitude_i': {'icon':'🌍 ', 'unit':''},
'longitude_i': {'icon':'', 'unit':''},
'altitude': {'icon':'⬆️ ', 'unit':'m'},
'location_source': {'icon':'📍 ', 'unit':''},
'altitude_source': {'icon':'⬆️ Source:', 'unit':''},
'ground_speed': {'icon':'💨 ', 'unit':'m/s'},
'ground_track': {'icon':'🧭 ', 'unit':'°'},
'PDOP': {'icon':'🎯 PDOP:', 'unit':''},
'HDOP': {'icon':'🎯 HDOP:', 'unit':''},
'VDOP': {'icon':'🎯 VDOP:', 'unit':''},
'sats_in_view': {'icon':'🛰️ ', 'unit':''},
'time': {'icon':'🕔 ', 'unit':''}
}
@@ -55,18 +75,20 @@ def get_chunks(data):
parsed=""
for item in reading:
key, value = item.split(":")
key, value = item.split(":", 1)
key = key.strip()
value = value.strip()
# If value is float, round it to the 1 digit after point
# else make it int
if "." in value:
value = round(float(value.strip()),1)
else:
try:
value = int(value)
except ValueError:
try:
value = int(value.strip())
except Exception:
# Leave it string as last resort
value = value
value = round(float(value), 1)
except ValueError:
# Leave nonnumeric values intact, including strings with periods.
pass
# Python 3.9-compatible alternative to match/case.
if key == "uptime_seconds":
@@ -79,6 +101,10 @@ def get_chunks(data):
elif key == "wind_direction":
# Convert wind direction from degrees to abbreviation
value = humanize_wind_direction(value)
elif key in ("PDOP", "HDOP", "VDOP"):
value = round(value / 100, 2)
elif key == "ground_track":
value = round(value / 100, 2)
elif key == "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
+3
View File
@@ -296,6 +296,9 @@ def parse_protobuf(packet: dict) -> Union[str, dict]:
if hasattr(pb, "environment_metrics") and pb.HasField("environment_metrics"):
return tb.get_chunks(str(pb.environment_metrics))
if hasattr(pb, "local_stats") and pb.HasField("local_stats"):
return tb.get_chunks(str(pb.local_stats))
# For other data, without implemented beautification, fallback to just printing the object
return str(pb).replace("\n", " ").replace("\r", " ").strip()
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.6.7"
version = "1.6.12"
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"}
+60 -1
View File
@@ -1,4 +1,5 @@
import unittest
from contextlib import ExitStack
from types import SimpleNamespace
from unittest import mock
@@ -11,6 +12,52 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
class ContactUiTests(unittest.TestCase):
def test_f5_restores_background_when_switching_between_different_size_nodes(self):
ui_state.node_list = [101, 102]
ui_state.current_window = 2
interface = SimpleNamespace(nodesByNum={
101: {"user": {"shortName": "BIG", "longName": "A" * 60}, "position": {"altitude": 100}},
102: {"user": {"shortName": "SMALL", "longName": "B"}},
})
win = mock.Mock()
win.getch.side_effect = [contact_ui.curses.KEY_DOWN, contact_ui.curses.KEY_UP, 27]
events = mock.Mock()
events.attach_mock(win.noutrefresh, "dialog")
stdscr = mock.Mock()
with ExitStack() as stack:
stack.enter_context(mock.patch.object(contact_ui.interface_state, "interface", interface))
for name, value in (("LINES", 40), ("COLS", 100)):
stack.enter_context(mock.patch.object(contact_ui.curses, name, value, create=True))
for name in ("curs_set", "update_lines_cols", "doupdate"):
stack.enter_context(mock.patch.object(contact_ui.curses, name))
stack.enter_context(mock.patch.object(contact_ui.curses, "newwin", return_value=win))
for name in ("get_color", "refresh_node_selection", "handle_resize"):
stack.enter_context(mock.patch.object(contact_ui, name, return_value=0))
restore = stack.enter_context(mock.patch.object(contact_ui, "stage_node_details_background"))
events.attach_mock(restore, "restore")
contact_ui.handle_f5_key(stdscr)
self.assertEqual(events.mock_calls, [
mock.call.dialog(), mock.call.restore(stdscr, 2), mock.call.dialog(),
mock.call.restore(stdscr, 2), mock.call.dialog(),
])
self.assertNotEqual(win.resize.call_args_list[0], win.resize.call_args_list[1])
self.assertEqual(win.refresh.call_count, 1) # Only on close.
def test_background_restore_stages_panes_without_flushing(self):
ui_state.current_window = 4
ui_state.single_pane_mode = True
stdscr = mock.Mock()
with ExitStack() as stack:
for name in ("channel_win", "messages_win", "nodes_win", "entry_win"):
stack.enter_context(mock.patch.object(contact_ui, name, mock.Mock(), create=True))
refresh = stack.enter_context(mock.patch.object(contact_ui, "refresh_pad"))
update = stack.enter_context(mock.patch.object(contact_ui.curses, "doupdate"))
contact_ui.stage_node_details_background(stdscr, 2)
refresh.assert_called_once_with(2, commit=False)
update.assert_not_called()
stdscr.noutrefresh.assert_called_once()
self.assertEqual(ui_state.current_window, 4)
def setUp(self) -> None:
reset_singletons()
self.saved_config = snapshot_config("single_pane_mode")
@@ -437,6 +484,8 @@ class ContactUiTests(unittest.TestCase):
self.assertEqual(contact_ui.get_window_title(1), "Primary")
def test_refresh_pad_draws_selected_channel_title_on_message_frame(self) -> None:
self.addCleanup(mock.patch.stopall)
mock.patch.object(contact_ui, "packetlog_win", mock.Mock(), create=True).start()
ui_state.single_pane_mode = True
ui_state.current_window = 1
ui_state.channel_list = ["Primary"]
@@ -453,9 +502,16 @@ class ContactUiTests(unittest.TestCase):
contact_ui.messages_win.getmaxyx.return_value = (10, 20)
with mock.patch.object(contact_ui, "get_msg_window_lines", return_value=4):
contact_ui.refresh_pad(1)
with mock.patch.object(contact_ui.curses, "doupdate") as update:
contact_ui.refresh_pad(1)
contact_ui.messages_win.addstr.assert_called_once_with(0, 2, " Primary ", contact_ui.curses.A_BOLD)
contact_ui.messages_win.refresh.assert_not_called()
contact_ui.messages_pad.refresh.assert_not_called()
contact_ui.messages_win.noutrefresh.assert_called_once()
contact_ui.messages_pad.touchwin.assert_called_once()
contact_ui.messages_pad.noutrefresh.assert_called_once()
update.assert_called_once()
def test_search_ignores_no_input_from_curses(self) -> None:
ui_state.node_list = [101]
@@ -508,6 +564,9 @@ class ContactUiTests(unittest.TestCase):
contact_ui.handle_f5_key(stdscr)
self.assertEqual(dialog_win.getch.call_count, 2)
# Idle polling must never present an erased dialog before its text.
self.assertEqual(dialog_win.noutrefresh.call_count, 2)
self.assertEqual(dialog_win.refresh.call_count, 1) # Closing the dialog only.
handle_resize.assert_called_once_with(stdscr, False)
def test_f5_node_details_tolerates_none_metrics(self) -> None:
+31
View File
@@ -1,4 +1,5 @@
from argparse import Namespace
from contextlib import ExitStack
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -10,6 +11,36 @@ from tests.test_support import reset_singletons
class ControlUiTests(unittest.TestCase):
def test_channel_list_renders_names_without_changing_slot_keys(self):
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
from contact.ui.menus import generate_menu_from_protobuf
from contact.utilities.singleton import menu_state
channels = [channel_pb2.Channel(settings=channel_pb2.ChannelSettings(name=name))
for name in ("Local", "Local", "", " ")]
node = SimpleNamespace(
localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig(),
getChannelByChannelIndex=lambda i: channels[i] if i < len(channels) else None,
)
menu_state.menu_path = ["Main Menu", "Channels"]
menu_state.current_menu = generate_menu_from_protobuf(None, node=node)["Main Menu"]["Channels"]
win, pad = mock.Mock(), mock.Mock()
win.getbegyx.return_value = (2, 2)
win.getmaxyx.return_value = (10, 80)
with ExitStack() as stack:
for name, value in (("LINES", 24), ("COLS", 100)):
stack.enter_context(mock.patch.object(control_ui.curses, name, value, create=True))
stack.enter_context(mock.patch.object(control_ui.curses, "newwin", return_value=win))
stack.enter_context(mock.patch.object(control_ui.curses, "newpad", return_value=pad))
stack.enter_context(mock.patch.object(control_ui.curses, "curs_set"))
stack.enter_context(mock.patch.object(control_ui, "get_menu_width", return_value=80))
for name in ("get_color", "draw_help_window", "draw_arrows"):
stack.enter_context(mock.patch.object(control_ui, name, return_value=0))
control_ui.display_menu()
labels = [call.args[2].strip() for call in pad.addstr.call_args_list]
self.assertEqual(labels, ["Local", "Local", "Channel 3", "Channel 4"])
self.assertEqual(list(menu_state.current_menu), [f"Channel {i}" for i in range(1, 5)])
def setUp(self) -> None:
reset_singletons()
+146
View File
@@ -0,0 +1,146 @@
from contextlib import ExitStack
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
from meshtastic.protobuf import admin_pb2, config_pb2, module_config_pb2
from contact.ui import control_ui
from contact.ui.menus import generate_menu_from_protobuf, update_ham_fields
from contact.utilities.save_to_radio import save_changes
from tests.test_support import reset_singletons
class HamSettingsTests(unittest.TestCase):
def build_interface(self, licensed=False):
node = Mock(nodeNum=2, localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig())
node.localConfig.lora.tx_power = 20
node.localConfig.lora.override_frequency = 915.5
node.getChannelByChannelIndex.return_value = None
info = {'user': {'longName': 'KD2ABC', 'shortName': 'HAM', 'isLicensed': licensed}}
interface = SimpleNamespace(localNode=node, getMyNodeInfo=lambda: info, nodesByNum={2: info})
node.iface = interface
return interface, node
def test_fields_follow_license_selection_and_installed_descriptors(self):
interface, node = self.build_interface()
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
descriptors = admin_pb2.HamParameters.DESCRIPTOR.fields
self.assertTrue(all(field.name not in user for field in descriptors))
user['isLicensed'] = (None, True)
update_ham_fields(user, node)
for field in descriptors:
self.assertIs(user[field.name][0], field)
self.assertEqual(user['call_sign'][1], 'KD2ABC')
self.assertEqual(user['tx_power'][1], 20)
self.assertEqual(user['frequency'][1], 915.5)
self.assertNotIn('longName', user)
user['frequency'] = (user['frequency'][0], 916.0)
update_ham_fields(user, node)
self.assertEqual(user['frequency'][1], 916.0)
user['isLicensed'] = (None, False)
update_ham_fields(user, node)
self.assertTrue(all(field.name not in user for field in descriptors))
self.assertEqual(user['longName'][1], 'KD2ABC')
self.assertEqual(user['shortName'][1], 'HAM')
def test_remote_menu_uses_remote_license_and_radio_values(self):
interface, node = self.build_interface(True)
interface.localNode = Mock()
user = generate_menu_from_protobuf(interface, node=node)['Main Menu']['User Settings']
self.assertEqual(user['frequency'][1], 915.5)
self.assertEqual(user['call_sign'][1], 'KD2ABC')
def test_save_full_ham_message_on_selected_node(self):
for remote in (False, True):
interface, node = self.build_interface(True)
if remote:
interface.localNode = Mock()
user = generate_menu_from_protobuf(interface, node=node)['Main Menu']['User Settings']
changes = {name: value for name, (_, value) in user.items()}
changes['frequency'] = 916.0
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, changes, state, node=node))
message = node._sendAdmin.call_args.args[0]
self.assertEqual(message.WhichOneof('payload_variant'), 'set_ham_mode')
self.assertEqual(message.set_ham_mode.call_sign, 'KD2ABC')
self.assertEqual(message.set_ham_mode.short_name, 'HAM')
self.assertEqual(message.set_ham_mode.tx_power, 20)
self.assertEqual(message.set_ham_mode.frequency, 916.0)
self.assertEqual(node._sendAdmin.call_args.kwargs['onResponse'], node.onAckNak if remote else None)
self.assertEqual([c[0] for c in node.method_calls], [
*['getChannelByChannelIndex'] * 8, 'ensureSessionKey', 'beginSettingsTransaction',
'setOwner', '_sendAdmin', 'commitSettingsTransaction'])
def test_disabling_license_does_not_send_ham_command(self):
interface, node = self.build_interface(True)
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
user['isLicensed'] = (None, False)
update_ham_fields(user, node)
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, {k: v[1] for k, v in user.items()}, state, node=node))
node._sendAdmin.assert_not_called()
node.setOwner.assert_called_once_with('KD2ABC', 'HAM', False, False)
def test_edit_through_normal_menu_after_selecting_licensed(self):
reset_singletons()
interface, node = self.build_interface()
win = Mock()
# Open User Settings, select licensed, edit the now-visible callsign, save.
win.getch.side_effect = [10, control_ui.curses.KEY_DOWN, control_ui.curses.KEY_DOWN,
10, control_ui.curses.KEY_DOWN, control_ui.curses.KEY_DOWN, 10, 9, 10, 27]
saved = {}
def capture(_interface, values, _state, **kwargs):
saved.update(values)
return False
try:
with ExitStack() as stack:
stack.enter_context(patch.object(control_ui.curses, 'update_lines_cols'))
stack.enter_context(patch.object(control_ui, 'display_menu', return_value=(win, Mock())))
stack.enter_context(patch.object(control_ui, 'move_highlight'))
stack.enter_context(patch.object(control_ui, 'get_list_input', return_value='True'))
text = stack.enter_context(patch.object(control_ui, 'get_text_input', return_value='KD2XYZ'))
stack.enter_context(patch.object(control_ui, 'save_changes', side_effect=capture))
control_ui.settings_menu(Mock(), interface)
self.assertEqual(text.call_args.args[1], 'call_sign')
self.assertEqual(saved['call_sign'], 'KD2XYZ')
self.assertTrue(saved['isLicensed'])
self.assertEqual(saved['tx_power'], 20)
finally:
reset_singletons()
def test_new_library_fields_are_discovered_edited_and_saved(self):
from google.protobuf import descriptor_pb2, descriptor_pool, message_factory
schema = descriptor_pb2.FileDescriptorProto(name='future_ham.proto', syntax='proto3')
params = schema.message_type.add(name='HamParameters')
for field in admin_pb2.HamParameters.DESCRIPTOR.fields:
params.field.add(name=field.name, number=field.number, type=field.type)
if 'long_name' not in admin_pb2.HamParameters.DESCRIPTOR.fields_by_name:
params.field.add(name='long_name', number=5, type=9)
params.field.add(name='future_option', number=100, type=8)
message = schema.message_type.add(name='AdminMessage')
message.field.add(name='set_ham_mode', number=18, type=11, type_name='.HamParameters')
pool = descriptor_pool.DescriptorPool()
pool.Add(schema)
future_params = message_factory.GetMessageClass(pool.FindMessageTypeByName('HamParameters'))
future_message = message_factory.GetMessageClass(pool.FindMessageTypeByName('AdminMessage'))
interface, node = self.build_interface(True)
with patch.object(admin_pb2, 'HamParameters', future_params), patch.object(admin_pb2, 'AdminMessage', future_message):
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
self.assertIn('long_name', user)
self.assertEqual(user['future_option'][1], False)
changes = {name: setting[1] for name, setting in user.items()}
changes.update(long_name='Alice', future_option=True)
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, changes, state, node=node))
sent = node._sendAdmin.call_args.args[0]
parsed = future_message.FromString(sent.SerializeToString())
self.assertEqual(parsed.set_ham_mode.long_name, 'Alice')
self.assertTrue(parsed.set_ham_mode.future_option)
def test_remote_user_settings_fetch_lora_before_showing_ham_defaults(self):
_, node = self.build_interface(True)
with patch.object(control_ui, '_request_remote_with_timeout') as request:
self.assertTrue(control_ui._request_remote_section(node, ['Main Menu'], 'User Settings'))
request.assert_called_once_with(node.requestConfig,
node.localConfig.DESCRIPTOR.fields_by_name['lora'],
cancel_callback=None)
+55
View File
@@ -0,0 +1,55 @@
from contextlib import ExitStack
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
import contact.ui.default_config
from contact.utilities import input_handlers
from tests.test_support import reset_singletons
class RestoredSettingsTests(unittest.TestCase):
def test_blank_input_is_unset_only_when_allowed(self):
for keys, allowed, expected in [
(["\n"], True, ""), ([" ", " ", "\n"], True, ""),
(["\x1b"], True, None), (["\n", "\x1b"], False, None),
]:
with self.subTest(keys=keys, allowed=allowed), ExitStack() as stack:
reset_singletons()
stack.callback(reset_singletons)
win = mock.Mock()
win.get_wch.side_effect = keys
stack.enter_context(mock.patch.object(input_handlers.curses, "newwin", return_value=win))
for name, value in (("LINES", 24), ("COLS", 80)):
stack.enter_context(mock.patch.object(input_handlers.curses, name, value, create=True))
stack.enter_context(mock.patch.object(input_handlers.curses, "curs_set"))
stack.enter_context(mock.patch.object(input_handlers, "get_color", return_value=0))
invalid = stack.enter_context(mock.patch.object(input_handlers, "invalid_input"))
result = input_handlers.get_text_input("Channel name", "name", str, allow_empty=allowed)
self.assertEqual(result, expected)
self.assertEqual(invalid.call_count, 0 if allowed else 1)
def test_missing_log_path_shows_configuration_guidance(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
config_path = root / "config.json"
log_path = root / "missing" / "client.log"
for module in ("contact", "contact.settings"):
with self.subTest(module=module):
config_path.write_text(json.dumps({"log_file_path": str(log_path)}))
result = subprocess.run(
[sys.executable, "-m", module],
env={**os.environ, "CONTACT_CONFIG_ROOT": str(root)},
cwd=Path(__file__).resolve().parents[1],
capture_output=True, text=True, timeout=15,
)
self.assertEqual(result.returncode, 1)
self.assertIn(str(log_path), result.stderr)
self.assertIn(f"Please edit config.json at: {config_path}", result.stderr)
self.assertIn('"log_file_path"', result.stderr)
self.assertNotIn("Traceback", result.stdout + result.stderr)
+17
View File
@@ -1,11 +1,28 @@
from types import SimpleNamespace
import unittest
from unittest import mock
from meshtastic.protobuf import channel_pb2
from contact.utilities.save_to_radio import save_changes
class SaveToRadioTests(unittest.TestCase):
def test_channel_enable_disable_and_unrelated_edits_preserve_role(self):
for index in (0, 1):
interface, node = self.build_interface()
node.channels = [channel_pb2.Channel(index=i) for i in range(2)]
channel = node.channels[index]
path = SimpleNamespace(menu_path=["Main Menu", "Channels", f"Channel {index + 1}"])
save_changes(interface, {"name": "Test"}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
save_changes(interface, {"enabled": True}, path)
expected = channel_pb2.Channel.PRIMARY if index == 0 else channel_pb2.Channel.SECONDARY
self.assertEqual(channel.role, expected)
save_changes(interface, {"enabled": False}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
self.assertEqual(channel.settings.name, "Test")
self.assertEqual(node.writeChannel.call_args_list, [mock.call(index)] * 3)
def build_interface(self):
node = mock.Mock()
node.localConfig = SimpleNamespace(
+6
View File
@@ -25,3 +25,9 @@ class TelemetryBeautifierTests(unittest.TestCase):
rendered = get_chunks("time:0\n")
self.assertIn("🕔 01.01.1970 00:00", rendered)
def test_get_chunks_normalizes_keys_and_preserves_unknown_strings(self) -> None:
rendered = get_chunks(' uptime_seconds: 7200\nunknown: "v1.2:ok"\n')
self.assertIn("🆙 2.0h", rendered)
self.assertIn('unknown:"v1.2:ok"', rendered)
+40 -1
View File
@@ -1,7 +1,7 @@
import unittest
from unittest import mock
from meshtastic.protobuf import config_pb2
from meshtastic.protobuf import config_pb2, mesh_pb2, telemetry_pb2
import contact.ui.default_config as config
from contact.utilities.demo_data import DEMO_LOCAL_NODE_NUM, build_demo_interface
@@ -19,6 +19,45 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
class UtilsTests(unittest.TestCase):
def test_parse_protobuf_beautifies_local_stats(self) -> None:
telemetry = telemetry_pb2.Telemetry(time=1788677531)
telemetry.local_stats.uptime_seconds = 7200
telemetry.local_stats.channel_utilization = 2.90333319
telemetry.local_stats.num_packets_tx = 12
telemetry.local_stats.num_packets_rx_bad = 3
telemetry.local_stats.num_online_nodes = 4
telemetry.local_stats.heap_free_bytes = 1024
rendered = parse_protobuf({"decoded": {
"portnum": "TELEMETRY_APP", "payload": telemetry.SerializeToString(),
}})
self.assertIn("🆙 2.0h", rendered)
self.assertIn("ChUtil:2.9%", rendered)
self.assertIn("📤 Tx:12", rendered)
self.assertIn("⚠️ RxBad:3", rendered)
self.assertIn("📡 Online:4", rendered)
self.assertIn("💾 Free:1024B", rendered)
self.assertNotIn("local_stats", rendered)
def test_parse_protobuf_beautifies_position_details(self) -> None:
position = mesh_pb2.Position(
latitude_i=454230020, longitude_i=-1228472320,
location_source=mesh_pb2.Position.LOC_INTERNAL,
ground_speed=2, ground_track=9000, PDOP=211, sats_in_view=8,
)
rendered = parse_protobuf({"decoded": {
"portnum": "POSITION_APP", "payload": position.SerializeToString(),
}})
self.assertIn("🌍 45.423002", rendered)
self.assertIn("-122.847232", rendered)
self.assertIn("📍 LOC_INTERNAL", rendered)
self.assertIn("💨 2m/s", rendered)
self.assertIn("🧭 90.0°", rendered)
self.assertIn("🎯 PDOP:2.11", rendered)
self.assertIn("🛰️ 8", rendered)
def setUp(self) -> None:
reset_singletons()
_get_modem_preset_name.cache_clear()