mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c7b54caf45 | ||
|
|
773f43edd8 | ||
|
|
6af1c46bd3 | ||
|
|
7e3e44df24 | ||
|
|
45626f5e83 | ||
|
|
e9181972b2 | ||
|
|
795ab84ef5 | ||
|
|
5e108c5fe5 | ||
|
|
edef37b116 | ||
|
|
e7e1bf7852 | ||
|
|
1c2384ea8d | ||
|
|
4cda264746 | ||
|
|
0005aaf438 | ||
|
|
f39a09646a |
@@ -5,9 +5,10 @@ import shutil
|
||||
import time
|
||||
import subprocess
|
||||
import threading
|
||||
from typing import Any, Dict, Optional
|
||||
# Debounce notification sounds so a burst of queued messages only plays once.
|
||||
_SOUND_DEBOUNCE_SECONDS = 0.8
|
||||
_sound_timer: threading.Timer | None = None
|
||||
_sound_timer: Optional[threading.Timer] = None
|
||||
_sound_timer_lock = threading.Lock()
|
||||
_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.daemon = True
|
||||
_sound_timer.start()
|
||||
from typing import Any, Dict
|
||||
|
||||
from contact.utilities.utils import (
|
||||
refresh_node_list,
|
||||
add_new_message,
|
||||
|
||||
@@ -19,6 +19,7 @@ from contact.utilities.singleton import ui_state, interface_state, menu_state
|
||||
|
||||
|
||||
MIN_COL = 1 # "effectively zero" without breaking curses
|
||||
RESIZE_DEBOUNCE_MS = 250
|
||||
root_win = None
|
||||
|
||||
|
||||
@@ -161,6 +162,24 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
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:
|
||||
"""Main UI loop for the curses interface."""
|
||||
global input_text
|
||||
@@ -168,6 +187,7 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
|
||||
root_win = stdscr
|
||||
input_text = ""
|
||||
queued_char = None
|
||||
stdscr.keypad(True)
|
||||
get_channels()
|
||||
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"))
|
||||
|
||||
# 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}")
|
||||
|
||||
@@ -224,7 +248,9 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
|
||||
elif char == curses.KEY_RESIZE:
|
||||
input_text = ""
|
||||
queued_char = drain_resize_events(entry_win)
|
||||
handle_resize(stdscr, False)
|
||||
continue
|
||||
|
||||
elif char == chr(4): # Ctrl + D to delete current channel or node
|
||||
handle_ctrl_d()
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import base64
|
||||
import curses
|
||||
import ipaddress
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
@@ -8,7 +9,9 @@ from typing import List
|
||||
from contact.utilities.save_to_radio import save_changes
|
||||
import contact.ui.default_config as config
|
||||
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 (
|
||||
get_repeated_input,
|
||||
get_text_input,
|
||||
@@ -16,7 +19,6 @@ from contact.utilities.input_handlers import (
|
||||
get_list_input,
|
||||
get_admin_key_input,
|
||||
)
|
||||
from contact.utilities.i18n import t
|
||||
from contact.ui.colors import get_color
|
||||
from contact.ui.dialog import dialog
|
||||
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
|
||||
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:
|
||||
global translation_file, field_mapping, help_text
|
||||
@@ -121,6 +130,22 @@ def display_menu() -> tuple[object, object]:
|
||||
full_key = ".".join(transformed_path + [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_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
|
||||
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 = current_value if new_value is None else new_value.split(", ")
|
||||
menu_state.start_index.pop()
|
||||
|
||||
@@ -6,7 +6,7 @@ from typing import Any, List, Dict, Optional
|
||||
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
|
||||
import contact.ui.default_config as config
|
||||
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.i18n import t
|
||||
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
|
||||
|
||||
|
||||
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:
|
||||
"""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)
|
||||
@@ -89,7 +100,7 @@ def setPref(config, comp_name, raw_val) -> bool:
|
||||
return False
|
||||
|
||||
# repeating fields need to be handled with append, not setattr
|
||||
if pref.label != pref.LABEL_REPEATED:
|
||||
if not _is_repeated_field(pref):
|
||||
try:
|
||||
if config_type.message_type is not None:
|
||||
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
|
||||
|
||||
|
||||
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]:
|
||||
"""Applies path replacements and normalizes entries in the menu path."""
|
||||
path_replacements = {"Radio Settings": "config", "Module Settings": "module"}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from typing import Optional
|
||||
|
||||
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 = {}
|
||||
_language = None
|
||||
|
||||
54
contact/utilities/ini_utils.py
Normal file
54
contact/utilities/ini_utils.py
Normal file
@@ -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:
|
||||
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
|
||||
width = get_dialog_width()
|
||||
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):
|
||||
curses.noecho()
|
||||
curses.curs_set(0)
|
||||
return int(ipaddress.ip_address(user_input))
|
||||
return int.from_bytes(ipaddress.IPv4Address(user_input).packed, "little", signed=False)
|
||||
else:
|
||||
fixed32_win.addstr(
|
||||
7,
|
||||
@@ -536,7 +539,7 @@ def get_fixed32_input(current_value: int) -> int:
|
||||
curses.napms(1500)
|
||||
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]
|
||||
|
||||
else:
|
||||
|
||||
@@ -112,16 +112,23 @@ def save_changes(interface, modified_settings, menu_state):
|
||||
else:
|
||||
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():
|
||||
# Check if the category exists in localConfig
|
||||
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
|
||||
config_subcategory = config_container
|
||||
|
||||
# Check if the config_item exists in the subcategory
|
||||
if hasattr(config_subcategory, config_item):
|
||||
|
||||
@@ -68,19 +68,19 @@ def get_chunks(data):
|
||||
# Leave it string as last resort
|
||||
value = value
|
||||
|
||||
match key:
|
||||
# Python 3.9-compatible alternative to match/case.
|
||||
if key == "uptime_seconds":
|
||||
# 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
|
||||
# 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
|
||||
case "wind_direction":
|
||||
value = humanize_wind_direction(value)
|
||||
case "time":
|
||||
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
|
||||
value = humanize_wind_direction(value)
|
||||
elif key == "time":
|
||||
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
|
||||
|
||||
if key in sensors:
|
||||
parsed+= f"{sensors[key.strip()]['icon']}{value}{sensors[key]['unit']} "
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.4.12"
|
||||
version = "1.4.18"
|
||||
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"}
|
||||
|
||||
Reference in New Issue
Block a user