mirror of
https://github.com/pdxlocations/contact.git
synced 2026-11-01 08:35:46 +01:00
Compare commits
4
Commits
1.6.12
...
role-unknown
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
afbde3c2d6 | ||
|
|
51351d316f | ||
|
|
888108f14e | ||
|
|
72f97bfde7 |
No files matched your search
@@ -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]
|
||||
|
||||
@@ -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')}")
|
||||
|
||||
|
||||
@@ -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
@@ -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"
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user