mirror of
https://github.com/pdxlocations/contact.git
synced 2026-11-09 10:28:46 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c5e93fdd2d | ||
|
|
35a0ff5834 | ||
|
|
18df7d326a | ||
|
|
c7b54caf45 | ||
|
|
773f43edd8 | ||
|
|
6af1c46bd3 | ||
|
|
7e3e44df24 | ||
|
|
45626f5e83 | ||
|
|
e9181972b2 | ||
|
|
795ab84ef5 | ||
|
|
5e108c5fe5 | ||
|
|
edef37b116 | ||
|
|
e7e1bf7852 | ||
|
|
1c2384ea8d | ||
|
|
4cda264746 | ||
|
|
0005aaf438 | ||
|
|
f39a09646a |
No files matched your search
@@ -5,9 +5,10 @@ import shutil
|
|||||||
import time
|
import time
|
||||||
import subprocess
|
import subprocess
|
||||||
import threading
|
import threading
|
||||||
|
from typing import Any, Dict, Optional
|
||||||
# Debounce notification sounds so a burst of queued messages only plays once.
|
# Debounce notification sounds so a burst of queued messages only plays once.
|
||||||
_SOUND_DEBOUNCE_SECONDS = 0.8
|
_SOUND_DEBOUNCE_SECONDS = 0.8
|
||||||
_sound_timer: threading.Timer | None = None
|
_sound_timer: Optional[threading.Timer] = None
|
||||||
_sound_timer_lock = threading.Lock()
|
_sound_timer_lock = threading.Lock()
|
||||||
_last_sound_request = 0.0
|
_last_sound_request = 0.0
|
||||||
|
|
||||||
@@ -42,8 +43,6 @@ def schedule_notification_sound(delay: float = _SOUND_DEBOUNCE_SECONDS) -> None:
|
|||||||
_sound_timer = threading.Timer(delay, _fire, args=(now,))
|
_sound_timer = threading.Timer(delay, _fire, args=(now,))
|
||||||
_sound_timer.daemon = True
|
_sound_timer.daemon = True
|
||||||
_sound_timer.start()
|
_sound_timer.start()
|
||||||
from typing import Any, Dict
|
|
||||||
|
|
||||||
from contact.utilities.utils import (
|
from contact.utilities.utils import (
|
||||||
refresh_node_list,
|
refresh_node_list,
|
||||||
add_new_message,
|
add_new_message,
|
||||||
|
|||||||
+45
-15
@@ -14,11 +14,12 @@ from contact.utilities.input_handlers import get_list_input
|
|||||||
from contact.utilities.i18n import t
|
from contact.utilities.i18n import t
|
||||||
import contact.ui.default_config as config
|
import contact.ui.default_config as config
|
||||||
import contact.ui.dialog
|
import contact.ui.dialog
|
||||||
from contact.ui.nav_utils import move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
|
from contact.ui.nav_utils import draw_main_arrows, fit_text, get_msg_window_lines, move_main_highlight, wrap_text
|
||||||
from contact.utilities.singleton import ui_state, interface_state, menu_state
|
from contact.utilities.singleton import ui_state, interface_state, menu_state
|
||||||
|
|
||||||
|
|
||||||
MIN_COL = 1 # "effectively zero" without breaking curses
|
MIN_COL = 1 # "effectively zero" without breaking curses
|
||||||
|
RESIZE_DEBOUNCE_MS = 250
|
||||||
root_win = None
|
root_win = None
|
||||||
|
|
||||||
|
|
||||||
@@ -161,6 +162,24 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def drain_resize_events(input_win: curses.window) -> Union[str, int, None]:
|
||||||
|
"""Wait for resize events to settle and preserve one queued non-resize key."""
|
||||||
|
input_win.timeout(RESIZE_DEBOUNCE_MS)
|
||||||
|
try:
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
next_char = input_win.get_wch()
|
||||||
|
except curses.error:
|
||||||
|
return None
|
||||||
|
|
||||||
|
if next_char == curses.KEY_RESIZE:
|
||||||
|
continue
|
||||||
|
|
||||||
|
return next_char
|
||||||
|
finally:
|
||||||
|
input_win.timeout(-1)
|
||||||
|
|
||||||
|
|
||||||
def main_ui(stdscr: curses.window) -> None:
|
def main_ui(stdscr: curses.window) -> None:
|
||||||
"""Main UI loop for the curses interface."""
|
"""Main UI loop for the curses interface."""
|
||||||
global input_text
|
global input_text
|
||||||
@@ -168,6 +187,7 @@ def main_ui(stdscr: curses.window) -> None:
|
|||||||
|
|
||||||
root_win = stdscr
|
root_win = stdscr
|
||||||
input_text = ""
|
input_text = ""
|
||||||
|
queued_char = None
|
||||||
stdscr.keypad(True)
|
stdscr.keypad(True)
|
||||||
get_channels()
|
get_channels()
|
||||||
handle_resize(stdscr, True)
|
handle_resize(stdscr, True)
|
||||||
@@ -176,7 +196,11 @@ def main_ui(stdscr: curses.window) -> None:
|
|||||||
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
|
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
|
||||||
|
|
||||||
# Get user input from entry window
|
# Get user input from entry window
|
||||||
char = entry_win.get_wch()
|
if queued_char is None:
|
||||||
|
char = entry_win.get_wch()
|
||||||
|
else:
|
||||||
|
char = queued_char
|
||||||
|
queued_char = None
|
||||||
|
|
||||||
# draw_debug(f"Keypress: {char}")
|
# draw_debug(f"Keypress: {char}")
|
||||||
|
|
||||||
@@ -224,7 +248,9 @@ def main_ui(stdscr: curses.window) -> None:
|
|||||||
|
|
||||||
elif char == curses.KEY_RESIZE:
|
elif char == curses.KEY_RESIZE:
|
||||||
input_text = ""
|
input_text = ""
|
||||||
|
queued_char = drain_resize_events(entry_win)
|
||||||
handle_resize(stdscr, False)
|
handle_resize(stdscr, False)
|
||||||
|
continue
|
||||||
|
|
||||||
elif char == chr(4): # Ctrl + D to delete current channel or node
|
elif char == chr(4): # Ctrl + D to delete current channel or node
|
||||||
handle_ctrl_d()
|
handle_ctrl_d()
|
||||||
@@ -802,9 +828,7 @@ def draw_channel_list() -> None:
|
|||||||
notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
|
notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
|
||||||
|
|
||||||
# Truncate the channel name if it's too long to fit in the window
|
# Truncate the channel name if it's too long to fit in the window
|
||||||
truncated_channel = (
|
truncated_channel = fit_text(f"{channel}{notification}", win_width - 3, suffix="-")
|
||||||
(channel[: win_width - 5] + "-" if len(channel) > win_width - 5 else channel) + notification
|
|
||||||
).ljust(win_width - 3)
|
|
||||||
|
|
||||||
color = get_color("channel_list")
|
color = get_color("channel_list")
|
||||||
if idx == ui_state.selected_channel:
|
if idx == ui_state.selected_channel:
|
||||||
@@ -910,8 +934,7 @@ def draw_node_list() -> None:
|
|||||||
snr_str = f" ■ SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
|
snr_str = f" ■ SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
|
||||||
|
|
||||||
# Future node name custom formatting possible
|
# Future node name custom formatting possible
|
||||||
node_str = f"{status_icon} {node_name}"
|
node_str = fit_text(f"{status_icon} {node_name}", box_width - 2)
|
||||||
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
|
|
||||||
color = "node_list"
|
color = "node_list"
|
||||||
if "isFavorite" in node and node["isFavorite"]:
|
if "isFavorite" in node and node["isFavorite"]:
|
||||||
color = "node_favorite"
|
color = "node_favorite"
|
||||||
@@ -1040,9 +1063,16 @@ def draw_packetlog_win() -> None:
|
|||||||
span += column
|
span += column
|
||||||
|
|
||||||
# Add headers
|
# Add headers
|
||||||
headers = f"{'From':<{columns[0]}} {'To':<{columns[1]}} {'Port':<{columns[2]}} {'Payload':<{width-span}}"
|
headers = " ".join(
|
||||||
|
[
|
||||||
|
fit_text("From", columns[0]),
|
||||||
|
fit_text("To", columns[1]),
|
||||||
|
fit_text("Port", columns[2]),
|
||||||
|
fit_text("Payload", max(1, width - span - 3)),
|
||||||
|
]
|
||||||
|
)
|
||||||
packetlog_win.addstr(
|
packetlog_win.addstr(
|
||||||
1, 1, headers[: width - 2], get_color("log_header", underline=True)
|
1, 1, fit_text(headers, width - 2), get_color("log_header", underline=True)
|
||||||
) # Truncate headers if they exceed window width
|
) # Truncate headers if they exceed window width
|
||||||
|
|
||||||
for i, packet in enumerate(reversed(ui_state.packet_buffer)):
|
for i, packet in enumerate(reversed(ui_state.packet_buffer)):
|
||||||
@@ -1050,22 +1080,22 @@ def draw_packetlog_win() -> None:
|
|||||||
break
|
break
|
||||||
|
|
||||||
# Format each field
|
# Format each field
|
||||||
from_id = get_name_from_database(packet["from"], "short").ljust(columns[0])
|
from_id = fit_text(get_name_from_database(packet["from"], "short"), columns[0])
|
||||||
to_id = (
|
to_id = (
|
||||||
"BROADCAST".ljust(columns[1])
|
fit_text("BROADCAST", columns[1])
|
||||||
if str(packet["to"]) == "4294967295"
|
if str(packet["to"]) == "4294967295"
|
||||||
else get_name_from_database(packet["to"], "short").ljust(columns[1])
|
else fit_text(get_name_from_database(packet["to"], "short"), columns[1])
|
||||||
)
|
)
|
||||||
if "decoded" in packet:
|
if "decoded" in packet:
|
||||||
port = str(packet["decoded"].get("portnum", "")).ljust(columns[2])
|
port = fit_text(str(packet["decoded"].get("portnum", "")), columns[2])
|
||||||
parsed_payload = parse_protobuf(packet)
|
parsed_payload = parse_protobuf(packet)
|
||||||
else:
|
else:
|
||||||
port = "NO KEY".ljust(columns[2])
|
port = fit_text("NO KEY", columns[2])
|
||||||
parsed_payload = "NO KEY"
|
parsed_payload = "NO KEY"
|
||||||
|
|
||||||
# Combine and truncate if necessary
|
# Combine and truncate if necessary
|
||||||
logString = f"{from_id} {to_id} {port} {parsed_payload}"
|
logString = f"{from_id} {to_id} {port} {parsed_payload}"
|
||||||
logString = logString[: width - 3]
|
logString = fit_text(logString, width - 3)
|
||||||
|
|
||||||
# Add to the window
|
# Add to the window
|
||||||
packetlog_win.addstr(i + 2, 1, logString, get_color("log"))
|
packetlog_win.addstr(i + 2, 1, logString, get_color("log"))
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import base64
|
import base64
|
||||||
import curses
|
import curses
|
||||||
|
import ipaddress
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
@@ -8,7 +9,9 @@ from typing import List
|
|||||||
from contact.utilities.save_to_radio import save_changes
|
from contact.utilities.save_to_radio import save_changes
|
||||||
import contact.ui.default_config as config
|
import contact.ui.default_config as config
|
||||||
from contact.utilities.config_io import config_export, config_import
|
from contact.utilities.config_io import config_export, config_import
|
||||||
from contact.utilities.control_utils import parse_ini_file, transform_menu_path
|
from contact.utilities.control_utils import transform_menu_path
|
||||||
|
from contact.utilities.i18n import t
|
||||||
|
from contact.utilities.ini_utils import parse_ini_file
|
||||||
from contact.utilities.input_handlers import (
|
from contact.utilities.input_handlers import (
|
||||||
get_repeated_input,
|
get_repeated_input,
|
||||||
get_text_input,
|
get_text_input,
|
||||||
@@ -16,7 +19,6 @@ from contact.utilities.input_handlers import (
|
|||||||
get_list_input,
|
get_list_input,
|
||||||
get_admin_key_input,
|
get_admin_key_input,
|
||||||
)
|
)
|
||||||
from contact.utilities.i18n import t
|
|
||||||
from contact.ui.colors import get_color
|
from contact.ui.colors import get_color
|
||||||
from contact.ui.dialog import dialog
|
from contact.ui.dialog import dialog
|
||||||
from contact.ui.menus import generate_menu_from_protobuf
|
from contact.ui.menus import generate_menu_from_protobuf
|
||||||
@@ -54,6 +56,13 @@ config_folder = os.path.abspath(config.node_configs_file_path)
|
|||||||
# Load translations
|
# Load translations
|
||||||
field_mapping, help_text = parse_ini_file(translation_file)
|
field_mapping, help_text = parse_ini_file(translation_file)
|
||||||
|
|
||||||
|
def _is_repeated_field(field_desc) -> bool:
|
||||||
|
"""Return True if the protobuf field is repeated.
|
||||||
|
Protobuf 6.31.0 and later use an is_repeated property, while older versions compare against the label field.
|
||||||
|
"""
|
||||||
|
if hasattr(field_desc, "is_repeated"):
|
||||||
|
return bool(field_desc.is_repeated)
|
||||||
|
return field_desc.label == field_desc.LABEL_REPEATED
|
||||||
|
|
||||||
def reload_translations() -> None:
|
def reload_translations() -> None:
|
||||||
global translation_file, field_mapping, help_text
|
global translation_file, field_mapping, help_text
|
||||||
@@ -121,6 +130,22 @@ def display_menu() -> tuple[object, object]:
|
|||||||
full_key = ".".join(transformed_path + [option])
|
full_key = ".".join(transformed_path + [option])
|
||||||
display_name = field_mapping.get(full_key, option)
|
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):
|
||||||
|
try:
|
||||||
|
current_value = str(
|
||||||
|
ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False))
|
||||||
|
)
|
||||||
|
except ipaddress.AddressValueError:
|
||||||
|
pass
|
||||||
|
elif isinstance(current_value, str) and current_value.isdigit():
|
||||||
|
try:
|
||||||
|
current_value = str(
|
||||||
|
ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False))
|
||||||
|
)
|
||||||
|
except ipaddress.AddressValueError:
|
||||||
|
pass
|
||||||
|
|
||||||
display_option = f"{display_name}"[: w // 2 - 2]
|
display_option = f"{display_name}"[: w // 2 - 2]
|
||||||
display_value = f"{current_value}"[: w // 2 - 4]
|
display_value = f"{current_value}"[: w // 2 - 4]
|
||||||
|
|
||||||
@@ -546,7 +571,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
|||||||
new_value = new_value == "True" or new_value is True
|
new_value = new_value == "True" or new_value is True
|
||||||
menu_state.start_index.pop()
|
menu_state.start_index.pop()
|
||||||
|
|
||||||
elif field.label == field.LABEL_REPEATED: # Handle repeated field - Not currently used
|
elif _is_repeated_field(field): # Handle repeated field - Not currently used
|
||||||
new_value = get_repeated_input(current_value)
|
new_value = get_repeated_input(current_value)
|
||||||
new_value = current_value if new_value is None else new_value.split(", ")
|
new_value = current_value if new_value is None else new_value.split(", ")
|
||||||
menu_state.start_index.pop()
|
menu_state.start_index.pop()
|
||||||
|
|||||||
+69
-16
@@ -1,11 +1,12 @@
|
|||||||
import curses
|
import curses
|
||||||
import re
|
import re
|
||||||
from unicodedata import east_asian_width
|
from typing import Any, Optional, List, Dict
|
||||||
|
|
||||||
|
from wcwidth import wcwidth, wcswidth
|
||||||
|
|
||||||
from contact.ui.colors import get_color
|
from contact.ui.colors import get_color
|
||||||
from contact.utilities.i18n import t
|
from contact.utilities.i18n import t
|
||||||
from contact.utilities.control_utils import transform_menu_path
|
from contact.utilities.control_utils import transform_menu_path
|
||||||
from typing import Any, Optional, List, Dict
|
|
||||||
from contact.utilities.singleton import interface_state, ui_state
|
from contact.utilities.singleton import interface_state, ui_state
|
||||||
|
|
||||||
|
|
||||||
@@ -328,7 +329,65 @@ def get_wrapped_help_text(
|
|||||||
|
|
||||||
|
|
||||||
def text_width(text: str) -> int:
|
def text_width(text: str) -> int:
|
||||||
return sum(2 if east_asian_width(c) in "FW" else 1 for c in text)
|
width = wcswidth(text)
|
||||||
|
if width >= 0:
|
||||||
|
return width
|
||||||
|
return sum(max(wcwidth(char), 0) for char in text)
|
||||||
|
|
||||||
|
|
||||||
|
def slice_text_to_width(text: str, max_width: int) -> str:
|
||||||
|
"""Return the longest prefix that fits within max_width terminal cells."""
|
||||||
|
if max_width <= 0:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
chunk = ""
|
||||||
|
for char in text:
|
||||||
|
candidate = chunk + char
|
||||||
|
if text_width(candidate) > max_width:
|
||||||
|
break
|
||||||
|
chunk = candidate
|
||||||
|
|
||||||
|
return chunk
|
||||||
|
|
||||||
|
|
||||||
|
def fit_text(text: str, width: int, suffix: str = "") -> str:
|
||||||
|
"""Trim and pad text so its terminal display width fits exactly."""
|
||||||
|
if width <= 0:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
if text_width(text) > width:
|
||||||
|
suffix = slice_text_to_width(suffix, width)
|
||||||
|
available = max(0, width - text_width(suffix))
|
||||||
|
text = slice_text_to_width(text, available).rstrip() + suffix
|
||||||
|
|
||||||
|
padding = max(0, width - text_width(text))
|
||||||
|
return text + (" " * padding)
|
||||||
|
|
||||||
|
|
||||||
|
def split_text_to_width(text: str, max_width: int) -> List[str]:
|
||||||
|
"""Split text into chunks that each fit within max_width terminal cells."""
|
||||||
|
if max_width <= 0:
|
||||||
|
return [""]
|
||||||
|
|
||||||
|
chunks: List[str] = []
|
||||||
|
chunk = ""
|
||||||
|
|
||||||
|
for char in text:
|
||||||
|
candidate = chunk + char
|
||||||
|
if chunk and text_width(candidate) > max_width:
|
||||||
|
chunks.append(chunk)
|
||||||
|
chunk = char
|
||||||
|
else:
|
||||||
|
chunk = candidate
|
||||||
|
|
||||||
|
if text_width(chunk) > max_width:
|
||||||
|
chunks.append(slice_text_to_width(chunk, max_width))
|
||||||
|
chunk = ""
|
||||||
|
|
||||||
|
if chunk:
|
||||||
|
chunks.append(chunk)
|
||||||
|
|
||||||
|
return chunks or [""]
|
||||||
|
|
||||||
|
|
||||||
def wrap_text(text: str, wrap_width: int) -> List[str]:
|
def wrap_text(text: str, wrap_width: int) -> List[str]:
|
||||||
@@ -346,24 +405,18 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
|
|||||||
wrap_width -= margin
|
wrap_width -= margin
|
||||||
|
|
||||||
for word in words:
|
for word in words:
|
||||||
word_length = text_width(word)
|
word_chunks = split_text_to_width(word, wrap_width) if text_width(word) > wrap_width else [word]
|
||||||
|
|
||||||
if word_length > wrap_width: # Break long words
|
for chunk in word_chunks:
|
||||||
if line_buffer:
|
chunk_length = text_width(chunk)
|
||||||
|
|
||||||
|
if line_length + chunk_length > wrap_width and chunk.strip():
|
||||||
wrapped_lines.append(line_buffer.strip())
|
wrapped_lines.append(line_buffer.strip())
|
||||||
line_buffer = ""
|
line_buffer = ""
|
||||||
line_length = 0
|
line_length = 0
|
||||||
for i in range(0, word_length, wrap_width):
|
|
||||||
wrapped_lines.append(word[i : i + wrap_width])
|
|
||||||
continue
|
|
||||||
|
|
||||||
if line_length + word_length > wrap_width and word.strip():
|
line_buffer += chunk
|
||||||
wrapped_lines.append(line_buffer.strip())
|
line_length += chunk_length
|
||||||
line_buffer = ""
|
|
||||||
line_length = 0
|
|
||||||
|
|
||||||
line_buffer += word
|
|
||||||
line_length += word_length
|
|
||||||
|
|
||||||
if line_buffer:
|
if line_buffer:
|
||||||
wrapped_lines.append(line_buffer.strip())
|
wrapped_lines.append(line_buffer.strip())
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from typing import Any, List, Dict, Optional
|
|||||||
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
|
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
|
||||||
import contact.ui.default_config as config
|
import contact.ui.default_config as config
|
||||||
from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
|
from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
|
||||||
from contact.utilities.control_utils import parse_ini_file
|
from contact.utilities.ini_utils import parse_ini_file
|
||||||
from contact.utilities.input_handlers import get_list_input
|
from contact.utilities.input_handlers import get_list_input
|
||||||
from contact.utilities.i18n import t
|
from contact.utilities.i18n import t
|
||||||
from contact.utilities.singleton import menu_state
|
from contact.utilities.singleton import menu_state
|
||||||
|
|||||||
@@ -9,6 +9,17 @@ from meshtastic.util import camel_to_snake, snake_to_camel, fromStr
|
|||||||
# defs are from meshtastic/python/main
|
# defs are from meshtastic/python/main
|
||||||
|
|
||||||
|
|
||||||
|
def _is_repeated_field(field_desc) -> bool:
|
||||||
|
"""Return True if the protobuf field is repeated.
|
||||||
|
|
||||||
|
Protobuf 6.31.0+ exposes `is_repeated`, while older versions require
|
||||||
|
checking `label == LABEL_REPEATED`.
|
||||||
|
"""
|
||||||
|
if hasattr(field_desc, "is_repeated"):
|
||||||
|
return bool(field_desc.is_repeated)
|
||||||
|
return field_desc.label == field_desc.LABEL_REPEATED
|
||||||
|
|
||||||
|
|
||||||
def traverseConfig(config_root, config, interface_config) -> bool:
|
def traverseConfig(config_root, config, interface_config) -> bool:
|
||||||
"""Iterate through current config level preferences and either traverse deeper if preference is a dict or set preference"""
|
"""Iterate through current config level preferences and either traverse deeper if preference is a dict or set preference"""
|
||||||
snake_name = camel_to_snake(config_root)
|
snake_name = camel_to_snake(config_root)
|
||||||
@@ -89,7 +100,7 @@ def setPref(config, comp_name, raw_val) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
# repeating fields need to be handled with append, not setattr
|
# repeating fields need to be handled with append, not setattr
|
||||||
if pref.label != pref.LABEL_REPEATED:
|
if not _is_repeated_field(pref):
|
||||||
try:
|
try:
|
||||||
if config_type.message_type is not None:
|
if config_type.message_type is not None:
|
||||||
config_values = getattr(config_part, config_type.name)
|
config_values = getattr(config_part, config_type.name)
|
||||||
|
|||||||
@@ -1,55 +1,7 @@
|
|||||||
from typing import Optional, Tuple, Dict, List
|
from typing import List
|
||||||
import re
|
import re
|
||||||
|
|
||||||
|
|
||||||
def parse_ini_file(ini_file_path: str) -> Tuple[Dict[str, str], Dict[str, str]]:
|
|
||||||
"""Parses an INI file and returns a mapping of keys to human-readable names and help text."""
|
|
||||||
|
|
||||||
field_mapping: Dict[str, str] = {}
|
|
||||||
help_text: Dict[str, str] = {}
|
|
||||||
current_section: Optional[str] = None
|
|
||||||
|
|
||||||
with open(ini_file_path, "r", encoding="utf-8") as f:
|
|
||||||
for line in f:
|
|
||||||
line = line.strip()
|
|
||||||
|
|
||||||
# Skip empty lines and comments
|
|
||||||
if not line or line.startswith(";") or line.startswith("#"):
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Handle sections like [config.device]
|
|
||||||
if line.startswith("[") and line.endswith("]"):
|
|
||||||
current_section = line[1:-1]
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Parse lines like: key, "Human-readable name", "helptext"
|
|
||||||
parts = [p.strip().strip('"') for p in line.split(",", 2)]
|
|
||||||
if len(parts) >= 2:
|
|
||||||
key = parts[0]
|
|
||||||
|
|
||||||
# If key is 'title', map directly to the section
|
|
||||||
if key == "title":
|
|
||||||
full_key = current_section
|
|
||||||
else:
|
|
||||||
full_key = f"{current_section}.{key}" if current_section else key
|
|
||||||
|
|
||||||
# Use the provided human-readable name or fallback to key
|
|
||||||
human_readable_name = parts[1] if parts[1] else key
|
|
||||||
field_mapping[full_key] = human_readable_name
|
|
||||||
|
|
||||||
# Handle help text or default
|
|
||||||
help = parts[2] if len(parts) == 3 and parts[2] else "No help available."
|
|
||||||
help_text[full_key] = help
|
|
||||||
|
|
||||||
else:
|
|
||||||
# Handle cases with only the key present
|
|
||||||
full_key = f"{current_section}.{key}" if current_section else key
|
|
||||||
field_mapping[full_key] = key
|
|
||||||
help_text[full_key] = "No help available."
|
|
||||||
|
|
||||||
return field_mapping, help_text
|
|
||||||
|
|
||||||
|
|
||||||
def transform_menu_path(menu_path: List[str]) -> List[str]:
|
def transform_menu_path(menu_path: List[str]) -> List[str]:
|
||||||
"""Applies path replacements and normalizes entries in the menu path."""
|
"""Applies path replacements and normalizes entries in the menu path."""
|
||||||
path_replacements = {"Radio Settings": "config", "Module Settings": "module"}
|
path_replacements = {"Radio Settings": "config", "Module Settings": "module"}
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import contact.ui.default_config as config
|
import contact.ui.default_config as config
|
||||||
from contact.utilities.control_utils import parse_ini_file
|
from contact.utilities.ini_utils import parse_ini_file
|
||||||
|
|
||||||
_translations = {}
|
_translations = {}
|
||||||
_language = None
|
_language = None
|
||||||
|
|||||||
@@ -0,0 +1,54 @@
|
|||||||
|
from typing import Optional, Tuple, Dict
|
||||||
|
from contact.utilities import i18n
|
||||||
|
|
||||||
|
|
||||||
|
def parse_ini_file(ini_file_path: str) -> Tuple[Dict[str, str], Dict[str, str]]:
|
||||||
|
"""Parses an INI file and returns a mapping of keys to human-readable names and help text."""
|
||||||
|
try:
|
||||||
|
default_help = i18n.t("ui.help.no_help", default="No help available.")
|
||||||
|
except Exception:
|
||||||
|
default_help = "No help available."
|
||||||
|
|
||||||
|
field_mapping: Dict[str, str] = {}
|
||||||
|
help_text: Dict[str, str] = {}
|
||||||
|
current_section: Optional[str] = None
|
||||||
|
|
||||||
|
with open(ini_file_path, "r", encoding="utf-8") as f:
|
||||||
|
for line in f:
|
||||||
|
line = line.strip()
|
||||||
|
|
||||||
|
# Skip empty lines and comments
|
||||||
|
if not line or line.startswith(";") or line.startswith("#"):
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Handle sections like [config.device]
|
||||||
|
if line.startswith("[") and line.endswith("]"):
|
||||||
|
current_section = line[1:-1]
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Parse lines like: key, "Human-readable name", "helptext"
|
||||||
|
parts = [p.strip().strip('"') for p in line.split(",", 2)]
|
||||||
|
if len(parts) >= 2:
|
||||||
|
key = parts[0]
|
||||||
|
|
||||||
|
# If key is 'title', map directly to the section
|
||||||
|
if key == "title":
|
||||||
|
full_key = current_section
|
||||||
|
else:
|
||||||
|
full_key = f"{current_section}.{key}" if current_section else key
|
||||||
|
|
||||||
|
# Use the provided human-readable name or fallback to key
|
||||||
|
human_readable_name = parts[1] if parts[1] else key
|
||||||
|
field_mapping[full_key] = human_readable_name
|
||||||
|
|
||||||
|
# Handle help text or default
|
||||||
|
help = parts[2] if len(parts) == 3 and parts[2] else default_help
|
||||||
|
help_text[full_key] = help
|
||||||
|
|
||||||
|
else:
|
||||||
|
# Handle cases with only the key present
|
||||||
|
full_key = f"{current_section}.{key}" if current_section else key
|
||||||
|
field_mapping[full_key] = key
|
||||||
|
help_text[full_key] = default_help
|
||||||
|
|
||||||
|
return field_mapping, help_text
|
||||||
@@ -462,7 +462,10 @@ from contact.utilities.singleton import menu_state # Ensure this is imported
|
|||||||
|
|
||||||
def get_fixed32_input(current_value: int) -> int:
|
def get_fixed32_input(current_value: int) -> int:
|
||||||
original_value = current_value
|
original_value = current_value
|
||||||
ip_string = str(ipaddress.IPv4Address(current_value))
|
try:
|
||||||
|
ip_string = str(ipaddress.IPv4Address(int(current_value).to_bytes(4, "little", signed=False)))
|
||||||
|
except Exception:
|
||||||
|
ip_string = str(ipaddress.IPv4Address(current_value))
|
||||||
height = 10
|
height = 10
|
||||||
width = get_dialog_width()
|
width = get_dialog_width()
|
||||||
start_y = max(0, (curses.LINES - height) // 2)
|
start_y = max(0, (curses.LINES - height) // 2)
|
||||||
@@ -524,7 +527,7 @@ def get_fixed32_input(current_value: int) -> int:
|
|||||||
if len(octets) == 4 and all(octet.isdigit() and 0 <= int(octet) <= 255 for octet in octets):
|
if len(octets) == 4 and all(octet.isdigit() and 0 <= int(octet) <= 255 for octet in octets):
|
||||||
curses.noecho()
|
curses.noecho()
|
||||||
curses.curs_set(0)
|
curses.curs_set(0)
|
||||||
return int(ipaddress.ip_address(user_input))
|
return int.from_bytes(ipaddress.IPv4Address(user_input).packed, "little", signed=False)
|
||||||
else:
|
else:
|
||||||
fixed32_win.addstr(
|
fixed32_win.addstr(
|
||||||
7,
|
7,
|
||||||
@@ -536,7 +539,7 @@ def get_fixed32_input(current_value: int) -> int:
|
|||||||
curses.napms(1500)
|
curses.napms(1500)
|
||||||
user_input = ""
|
user_input = ""
|
||||||
|
|
||||||
elif key in (curses.KEY_BACKSPACE, 127):
|
elif key in (curses.KEY_BACKSPACE, curses.KEY_DC, 127, 8, "\b", "\x7f"):
|
||||||
user_input = user_input[:-1]
|
user_input = user_input[:-1]
|
||||||
|
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -112,16 +112,23 @@ def save_changes(interface, modified_settings, menu_state):
|
|||||||
else:
|
else:
|
||||||
config_category = None
|
config_category = None
|
||||||
|
|
||||||
|
# Resolve the target config container, including nested sub-messages (e.g., network.ipv4_config)
|
||||||
|
config_container = None
|
||||||
|
if hasattr(node.localConfig, config_category):
|
||||||
|
config_container = getattr(node.localConfig, config_category)
|
||||||
|
elif hasattr(node.moduleConfig, config_category):
|
||||||
|
config_container = getattr(node.moduleConfig, config_category)
|
||||||
|
else:
|
||||||
|
logging.warning(f"Config category '{config_category}' not found in config.")
|
||||||
|
return
|
||||||
|
|
||||||
|
if len(menu_state.menu_path) >= 4:
|
||||||
|
nested_key = menu_state.menu_path[3]
|
||||||
|
if hasattr(config_container, nested_key):
|
||||||
|
config_container = getattr(config_container, nested_key)
|
||||||
|
|
||||||
for config_item, new_value in modified_settings.items():
|
for config_item, new_value in modified_settings.items():
|
||||||
# Check if the category exists in localConfig
|
config_subcategory = config_container
|
||||||
if hasattr(node.localConfig, config_category):
|
|
||||||
config_subcategory = getattr(node.localConfig, config_category)
|
|
||||||
# Check if the category exists in moduleConfig
|
|
||||||
elif hasattr(node.moduleConfig, config_category):
|
|
||||||
config_subcategory = getattr(node.moduleConfig, config_category)
|
|
||||||
else:
|
|
||||||
logging.warning(f"Config category '{config_category}' not found in config.")
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Check if the config_item exists in the subcategory
|
# Check if the config_item exists in the subcategory
|
||||||
if hasattr(config_subcategory, config_item):
|
if hasattr(config_subcategory, config_item):
|
||||||
|
|||||||
@@ -68,19 +68,19 @@ def get_chunks(data):
|
|||||||
# Leave it string as last resort
|
# Leave it string as last resort
|
||||||
value = value
|
value = value
|
||||||
|
|
||||||
match key:
|
# Python 3.9-compatible alternative to match/case.
|
||||||
|
if key == "uptime_seconds":
|
||||||
# convert seconds to hours, for our sanity
|
# convert seconds to hours, for our sanity
|
||||||
case "uptime_seconds":
|
value = round(value / 60 / 60, 1)
|
||||||
value = round(value / 60 / 60, 1)
|
elif key in ("longitude_i", "latitude_i"):
|
||||||
# Convert position to degrees (humanize), as per Meshtastic protobuf comment for this telemetry
|
# Convert position to degrees (humanize), as per Meshtastic protobuf comment for this telemetry
|
||||||
# truncate to 6th digit after floating point, which would be still accurate
|
# truncate to 6th digit after floating point, which would be still accurate
|
||||||
case "longitude_i" | "latitude_i":
|
value = round(value * 1e-7, 6)
|
||||||
value = round(value * 1e-7, 6)
|
elif key == "wind_direction":
|
||||||
# Convert wind direction from degrees to abbreviation
|
# Convert wind direction from degrees to abbreviation
|
||||||
case "wind_direction":
|
value = humanize_wind_direction(value)
|
||||||
value = humanize_wind_direction(value)
|
elif key == "time":
|
||||||
case "time":
|
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
|
||||||
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
|
|
||||||
|
|
||||||
if key in sensors:
|
if key in sensors:
|
||||||
parsed+= f"{sensors[key.strip()]['icon']}{value}{sensors[key]['unit']} "
|
parsed+= f"{sensors[key.strip()]['icon']}{value}{sensors[key]['unit']} "
|
||||||
|
|||||||
+3
-2
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "contact"
|
name = "contact"
|
||||||
version = "1.4.12"
|
version = "1.4.19"
|
||||||
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."
|
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 = [
|
authors = [
|
||||||
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
|
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
|
||||||
@@ -9,7 +9,8 @@ license = "GPL-3.0-only"
|
|||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.9,<3.15"
|
requires-python = ">=3.9,<3.15"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"meshtastic (>=2.7.5,<3.0.0)"
|
"meshtastic (>=2.7.5,<3.0.0)",
|
||||||
|
"wcwidth (>=0.2.13,<1.0.0)"
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.urls]
|
[project.urls]
|
||||||
|
|||||||
@@ -1 +1,2 @@
|
|||||||
meshtastic
|
meshtastic
|
||||||
|
wcwidth>=0.2.13,<1.0.0
|
||||||
Reference in new issue
Block a user