Compare commits

...
4 Commits
10 changed files with 308 additions and 17 deletions

No files matched your search

+6 -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]
+4 -1
View File
@@ -971,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')}")
+19 -4
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
@@ -360,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
@@ -508,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:
@@ -762,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"]:
@@ -832,6 +838,9 @@ 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"
@@ -863,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:
+34 -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"
+16 -1
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(
+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()
+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)
+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()