mirror of
https://github.com/pdxlocations/contact.git
synced 2026-10-31 16:21:57 +01:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
808fee4fe0 | ||
|
|
94d0a9053e | ||
|
|
f38167db1a | ||
|
|
57caf463bc |
No files matched your search
+4
-2
@@ -121,7 +121,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 +162,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)
|
||||
|
||||
|
||||
|
||||
@@ -160,7 +160,6 @@ 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."
|
||||
|
||||
+1
-1
@@ -36,7 +36,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(
|
||||
|
||||
@@ -97,17 +97,6 @@ def get_translated_header(menu_path: List[str]) -> str:
|
||||
return f"{admin_target_label} | {breadcrumb}" if admin_target_label else breadcrumb
|
||||
|
||||
|
||||
def get_settings_option_label(option, field_info, menu_path):
|
||||
if menu_path == ["Main Menu", "Channels"] and isinstance(field_info, dict):
|
||||
name = field_info.get("name", (None, ""))[1]
|
||||
if name:
|
||||
return name
|
||||
full_key = ".".join(transform_menu_path(menu_path) + [option])
|
||||
if len(menu_path) == 3 and menu_path[1] == "Channels" and option == "enabled":
|
||||
return field_mapping.get(full_key, "Enabled")
|
||||
return field_mapping.get(full_key, option)
|
||||
|
||||
|
||||
def display_menu() -> tuple[object, object]:
|
||||
# if help_win:
|
||||
# min_help_window_height = 6
|
||||
@@ -151,7 +140,7 @@ def display_menu() -> tuple[object, object]:
|
||||
field_info = menu_state.current_menu[option]
|
||||
current_value = field_info[1] if isinstance(field_info, tuple) else ""
|
||||
full_key = ".".join(transformed_path + [option])
|
||||
display_name = get_settings_option_label(option, field_info, menu_state.menu_path)
|
||||
display_name = field_mapping.get(full_key, option)
|
||||
|
||||
if full_key.startswith("config.network.ipv4_config.") and option in {"ip", "gateway", "subnet", "dns"}:
|
||||
if isinstance(current_value, int):
|
||||
@@ -730,7 +719,7 @@ def settings_menu(
|
||||
full_key = ".".join(transformed_path + [selected_option])
|
||||
|
||||
# Fetch human-readable name from field_mapping
|
||||
human_readable_name = get_settings_option_label(selected_option, field_info, menu_state.menu_path)
|
||||
human_readable_name = field_mapping.get(full_key, selected_option)
|
||||
|
||||
if field is None and selected_option in ("ringtone", "messages"):
|
||||
is_ringtone = selected_option == "ringtone"
|
||||
@@ -790,11 +779,6 @@ 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":
|
||||
@@ -836,15 +820,8 @@ 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"
|
||||
and menu_state.menu_path[2] != "Channel 1"
|
||||
)
|
||||
new_value = get_text_input(
|
||||
f"{human_readable_name} is currently: {current_value}", selected_option, input_type,
|
||||
allow_empty=allow_empty,
|
||||
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
|
||||
)
|
||||
new_value = current_value if new_value is None else new_value
|
||||
menu_state.start_index.pop()
|
||||
|
||||
@@ -93,10 +93,6 @@ 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
@@ -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()
|
||||
|
||||
@@ -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, *, allow_empty: bool = False) -> Optional[str]:
|
||||
def get_text_input(prompt: str, selected_config: str, input_type: str) -> Optional[str]:
|
||||
"""Handles user input with wrapped text for long prompts."""
|
||||
|
||||
def redraw_input_win():
|
||||
@@ -127,9 +127,6 @@ def get_text_input(prompt: str, selected_config: str, input_type: str, *, allow_
|
||||
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."),
|
||||
|
||||
@@ -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}")
|
||||
|
||||
@@ -160,15 +160,7 @@ def save_changes(interface, modified_settings, menu_state, node=None):
|
||||
|
||||
channel = node.channels[channel_num]
|
||||
for key, value in modified_settings.items():
|
||||
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
|
||||
if 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,
|
||||
@@ -177,6 +169,11 @@ 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.7"
|
||||
version = "1.6.9"
|
||||
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
-19
@@ -1,30 +1,12 @@
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
|
||||
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
|
||||
from meshtastic.protobuf import config_pb2, module_config_pb2
|
||||
|
||||
from contact.ui.menus import generate_menu_from_protobuf
|
||||
|
||||
|
||||
class MenusTests(unittest.TestCase):
|
||||
def test_channels_include_enabled_state_and_display_names(self):
|
||||
from contact.ui.control_ui import get_settings_option_label
|
||||
|
||||
channels = [
|
||||
channel_pb2.Channel(role=channel_pb2.Channel.PRIMARY, settings=channel_pb2.ChannelSettings(name="Local")),
|
||||
channel_pb2.Channel(role=channel_pb2.Channel.DISABLED, settings=channel_pb2.ChannelSettings(name="Local")),
|
||||
channel_pb2.Channel(role=channel_pb2.Channel.DISABLED),
|
||||
]
|
||||
node = SimpleNamespace(
|
||||
localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig(),
|
||||
getChannelByChannelIndex=lambda index: channels[index] if index < len(channels) else None,
|
||||
)
|
||||
menu = generate_menu_from_protobuf(None, node=node)["Main Menu"]["Channels"]
|
||||
self.assertEqual(menu["Channel 1"]["enabled"], (None, True))
|
||||
self.assertEqual(menu["Channel 2"]["enabled"], (None, False))
|
||||
labels = [get_settings_option_label(key, value, ["Main Menu", "Channels"]) for key, value in menu.items()]
|
||||
self.assertEqual(labels, ["Local", "Local", "Channel 3"])
|
||||
|
||||
def test_main_menu_includes_factory_reset_config_after_factory_reset(self) -> None:
|
||||
local_node = SimpleNamespace(
|
||||
localConfig=config_pb2.Config(),
|
||||
|
||||
@@ -1,28 +1,11 @@
|
||||
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