mirror of
https://github.com/pdxlocations/contact.git
synced 2026-11-01 08:35:46 +01:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
31bff496f1 | ||
|
|
b7c3ed20f2 | ||
|
|
7f7e0598d2 | ||
|
|
82e7456248 | ||
|
|
7b21c9c886 | ||
|
|
392d236a40 | ||
|
|
546fab099c |
No files matched your search
+2
-9
@@ -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()
|
||||
|
||||
|
||||
@@ -160,6 +160,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."
|
||||
|
||||
+2
-6
@@ -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:
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
@@ -779,6 +785,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 +831,14 @@ def settings_menu(
|
||||
|
||||
else: # Handle other field types
|
||||
input_type = get_input_type_for_field(field)
|
||||
allow_empty = (
|
||||
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()
|
||||
|
||||
@@ -93,6 +93,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
|
||||
|
||||
@@ -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."),
|
||||
|
||||
@@ -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
|
||||
@@ -160,7 +160,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 +177,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}")
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.6.9"
|
||||
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"}
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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(
|
||||
|
||||
Reference in new issue
Block a user