Compare commits

..

1 Commits

Author SHA1 Message Date
pdxlocations
ba713366bc bump version 2026-02-11 21:58:20 -08:00
7 changed files with 20 additions and 70 deletions

View File

@@ -5,10 +5,9 @@ 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: Optional[threading.Timer] = None
_sound_timer: threading.Timer | None = None
_sound_timer_lock = threading.Lock()
_last_sound_request = 0.0
@@ -43,6 +42,8 @@ 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,

View File

@@ -19,7 +19,6 @@ 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
@@ -162,24 +161,6 @@ 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
@@ -187,7 +168,6 @@ def main_ui(stdscr: curses.window) -> None:
root_win = stdscr
input_text = ""
queued_char = None
stdscr.keypad(True)
get_channels()
handle_resize(stdscr, True)
@@ -196,11 +176,7 @@ 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
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
char = entry_win.get_wch()
# draw_debug(f"Keypress: {char}")
@@ -248,9 +224,7 @@ 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()

View File

@@ -56,13 +56,6 @@ 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
@@ -133,16 +126,12 @@ def display_menu() -> tuple[object, object]:
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))
)
current_value = str(ipaddress.IPv4Address(current_value))
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))
)
current_value = str(ipaddress.IPv4Address(int(current_value)))
except ipaddress.AddressValueError:
pass
@@ -571,7 +560,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 _is_repeated_field(field): # Handle repeated field - Not currently used
elif field.label == field.LABEL_REPEATED: # 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()

View File

@@ -9,17 +9,6 @@ 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)
@@ -100,7 +89,7 @@ def setPref(config, comp_name, raw_val) -> bool:
return False
# repeating fields need to be handled with append, not setattr
if not _is_repeated_field(pref):
if pref.label != pref.LABEL_REPEATED:
try:
if config_type.message_type is not None:
config_values = getattr(config_part, config_type.name)

View File

@@ -462,10 +462,7 @@ from contact.utilities.singleton import menu_state # Ensure this is imported
def get_fixed32_input(current_value: int) -> int:
original_value = 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))
ip_string = str(ipaddress.IPv4Address(current_value))
height = 10
width = get_dialog_width()
start_y = max(0, (curses.LINES - height) // 2)
@@ -527,7 +524,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.from_bytes(ipaddress.IPv4Address(user_input).packed, "little", signed=False)
return int(ipaddress.ip_address(user_input))
else:
fixed32_win.addstr(
7,

View File

@@ -68,19 +68,19 @@ def get_chunks(data):
# Leave it string as last resort
value = value
# Python 3.9-compatible alternative to match/case.
if key == "uptime_seconds":
match key:
# convert seconds to hours, for our sanity
value = round(value / 60 / 60, 1)
elif key in ("longitude_i", "latitude_i"):
case "uptime_seconds":
value = round(value / 60 / 60, 1)
# 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
value = round(value * 1e-7, 6)
elif key == "wind_direction":
case "longitude_i" | "latitude_i":
value = round(value * 1e-7, 6)
# Convert wind direction from degrees to abbreviation
value = humanize_wind_direction(value)
elif key == "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
case "wind_direction":
value = humanize_wind_direction(value)
case "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']} "

View File

@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.4.18"
version = "1.4.14"
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"}