mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
working changes (#172)
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from typing import Any, Dict
|
||||
|
||||
from contact.utilities.utils import refresh_node_list
|
||||
from contact.ui.contact_ui import (
|
||||
@@ -21,7 +21,7 @@ import contact.ui.default_config as config
|
||||
import contact.globals as globals
|
||||
|
||||
|
||||
def on_receive(packet: dict[str, Any], interface: Any) -> None:
|
||||
def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
"""
|
||||
Handles an incoming packet from a Meshtastic interface.
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from typing import Any, Dict
|
||||
|
||||
import google.protobuf.json_format
|
||||
from meshtastic import BROADCAST_NUM
|
||||
@@ -15,12 +15,12 @@ from contact.utilities.db_handler import (
|
||||
import contact.ui.default_config as config
|
||||
import contact.globals as globals
|
||||
|
||||
ack_naks: dict[str, dict[str, Any]] = {} # requestId -> {channel, messageIndex, timestamp}
|
||||
ack_naks: Dict[str, Dict[str, Any]] = {} # requestId -> {channel, messageIndex, timestamp}
|
||||
|
||||
|
||||
# Note "onAckNak" has special meaning to the API, thus the nonstandard naming convention
|
||||
# See https://github.com/meshtastic/python/blob/master/meshtastic/mesh_interface.py#L462
|
||||
def onAckNak(packet: dict[str, Any]) -> None:
|
||||
def onAckNak(packet: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Handles incoming ACK/NAK response packets.
|
||||
"""
|
||||
@@ -58,7 +58,7 @@ def onAckNak(packet: dict[str, Any]) -> None:
|
||||
draw_messages_window()
|
||||
|
||||
|
||||
def on_response_traceroute(packet: dict[str, Any]) -> None:
|
||||
def on_response_traceroute(packet: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Handle traceroute response packets and render the route visually in the UI.
|
||||
"""
|
||||
|
||||
@@ -2,6 +2,7 @@ import curses
|
||||
import textwrap
|
||||
import logging
|
||||
import traceback
|
||||
from typing import Union
|
||||
|
||||
from contact.utilities.utils import get_channels, get_readable_duration, get_time_ago, refresh_node_list
|
||||
from contact.settings import settings_menu
|
||||
@@ -883,6 +884,6 @@ def draw_centered_text_field(win: curses.window, text: str, y_offset: int, color
|
||||
win.refresh()
|
||||
|
||||
|
||||
def draw_debug(value: str | int) -> None:
|
||||
def draw_debug(value: Union[str, int]) -> None:
|
||||
function_win.addstr(1, 1, f"debug: {value} ")
|
||||
function_win.refresh()
|
||||
|
||||
@@ -3,6 +3,7 @@ import curses
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from typing import List
|
||||
|
||||
from contact.utilities.save_to_radio import save_changes
|
||||
from contact.utilities.config_io import config_export, config_import
|
||||
@@ -130,7 +131,7 @@ def draw_help_window(
|
||||
menu_start_x: int,
|
||||
menu_height: int,
|
||||
max_help_lines: int,
|
||||
transformed_path: list[str],
|
||||
transformed_path: List[str],
|
||||
menu_state: MenuState,
|
||||
) -> None:
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Dict
|
||||
|
||||
# Get the parent directory of the script
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
@@ -12,7 +13,7 @@ log_file_path = os.path.join(parent_dir, "client.log")
|
||||
db_file_path = os.path.join(parent_dir, "client.db")
|
||||
|
||||
|
||||
def format_json_single_line_arrays(data: dict[str, object], indent: int = 4) -> str:
|
||||
def format_json_single_line_arrays(data: Dict[str, object], indent: int = 4) -> str:
|
||||
"""
|
||||
Formats JSON with arrays on a single line while keeping other elements properly indented.
|
||||
"""
|
||||
@@ -32,7 +33,7 @@ def format_json_single_line_arrays(data: dict[str, object], indent: int = 4) ->
|
||||
|
||||
|
||||
# Recursive function to check and update nested dictionaries
|
||||
def update_dict(default: dict[str, object], actual: dict[str, object]) -> bool:
|
||||
def update_dict(default: Dict[str, object], actual: Dict[str, object]) -> bool:
|
||||
updated = False
|
||||
for key, value in default.items():
|
||||
if key not in actual:
|
||||
@@ -44,7 +45,7 @@ def update_dict(default: dict[str, object], actual: dict[str, object]) -> bool:
|
||||
return updated
|
||||
|
||||
|
||||
def initialize_config() -> dict[str, object]:
|
||||
def initialize_config() -> Dict[str, object]:
|
||||
COLOR_CONFIG_DARK = {
|
||||
"default": ["white", "black"],
|
||||
"background": [" ", "black"],
|
||||
@@ -164,7 +165,7 @@ def initialize_config() -> dict[str, object]:
|
||||
return loaded_config
|
||||
|
||||
|
||||
def assign_config_variables(loaded_config: dict[str, object]) -> None:
|
||||
def assign_config_variables(loaded_config: Dict[str, object]) -> None:
|
||||
# Assign values to local variables
|
||||
|
||||
global db_file_path, log_file_path, message_prefix, sent_message_prefix
|
||||
|
||||
@@ -3,7 +3,7 @@ import logging
|
||||
import os
|
||||
from collections import OrderedDict
|
||||
|
||||
from typing import Any
|
||||
from typing import Any, Union, Dict
|
||||
|
||||
from google.protobuf.message import Message
|
||||
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
|
||||
@@ -21,8 +21,8 @@ def encode_if_bytes(value: Any) -> str:
|
||||
|
||||
|
||||
def extract_fields(
|
||||
message_instance: Message, current_config: Message | dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
message_instance: Message, current_config: Union[Message, Dict[str, Any], None] = None
|
||||
) -> Dict[str, Any]:
|
||||
if isinstance(current_config, dict): # Handle dictionaries
|
||||
return {key: (None, encode_if_bytes(current_config.get(key, "Not Set"))) for key in current_config}
|
||||
|
||||
@@ -63,7 +63,7 @@ def extract_fields(
|
||||
return menu
|
||||
|
||||
|
||||
def generate_menu_from_protobuf(interface: object) -> dict[str, Any]:
|
||||
def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
|
||||
"""
|
||||
Builds the full settings menu structure from the protobuf definitions.
|
||||
"""
|
||||
|
||||
@@ -2,11 +2,11 @@ import curses
|
||||
import re
|
||||
from contact.ui.colors import get_color
|
||||
from contact.utilities.control_utils import transform_menu_path
|
||||
from typing import Any
|
||||
from typing import Any, Optional, List, Dict
|
||||
|
||||
# Aliases
|
||||
Segment = tuple[str, str, bool, bool]
|
||||
WrappedLine = list[Segment]
|
||||
WrappedLine = List[Segment]
|
||||
|
||||
width = 80
|
||||
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
|
||||
@@ -14,7 +14,7 @@ save_option = "Save Changes"
|
||||
|
||||
|
||||
def move_highlight(
|
||||
old_idx: int, options: list[str], menu_win: curses.window, menu_pad: curses.window, **kwargs: Any
|
||||
old_idx: int, options: List[str], menu_win: curses.window, menu_pad: curses.window, **kwargs: Any
|
||||
) -> None:
|
||||
|
||||
show_save_option = None
|
||||
@@ -125,7 +125,7 @@ def move_highlight(
|
||||
|
||||
|
||||
def draw_arrows(
|
||||
win: object, visible_height: int, max_index: int, start_index: list[int], show_save_option: bool
|
||||
win: object, visible_height: int, max_index: int, start_index: List[int], show_save_option: bool
|
||||
) -> None:
|
||||
|
||||
# vh = visible_height + (1 if show_save_option else 0)
|
||||
@@ -145,9 +145,9 @@ def draw_arrows(
|
||||
|
||||
def update_help_window(
|
||||
help_win: object, # curses window or None
|
||||
help_text: dict[str, str],
|
||||
transformed_path: list[str],
|
||||
selected_option: str | None,
|
||||
help_text: Dict[str, str],
|
||||
transformed_path: List[str],
|
||||
selected_option: Optional[str],
|
||||
max_help_lines: int,
|
||||
width: int,
|
||||
help_y: int,
|
||||
@@ -191,8 +191,8 @@ def update_help_window(
|
||||
|
||||
|
||||
def get_wrapped_help_text(
|
||||
help_text: dict[str, str], transformed_path: list[str], selected_option: str | None, width: int, max_lines: int
|
||||
) -> list[WrappedLine]:
|
||||
help_text: Dict[str, str], transformed_path: List[str], selected_option: Optional[str], width: int, max_lines: int
|
||||
) -> List[WrappedLine]:
|
||||
"""Fetches and formats help text for display, ensuring it fits within the allowed lines."""
|
||||
|
||||
full_help_key = ".".join(transformed_path + [selected_option]) if selected_option else None
|
||||
@@ -210,7 +210,7 @@ def get_wrapped_help_text(
|
||||
r"\\033\[4m(.*?)\\033\[0m": ("settings_default", False, True), # Underline
|
||||
}
|
||||
|
||||
def extract_ansi_segments(text: str) -> list[Segment]:
|
||||
def extract_ansi_segments(text: str) -> List[Segment]:
|
||||
"""Extracts and replaces ANSI color codes, ensuring spaces are preserved."""
|
||||
matches = []
|
||||
last_pos = 0
|
||||
@@ -240,7 +240,7 @@ def get_wrapped_help_text(
|
||||
|
||||
return matches
|
||||
|
||||
def wrap_ansi_text(segments: list[Segment], wrap_width: int) -> list[WrappedLine]:
|
||||
def wrap_ansi_text(segments: List[Segment], wrap_width: int) -> List[WrappedLine]:
|
||||
"""Wraps text while preserving ANSI formatting and spaces."""
|
||||
wrapped_lines = []
|
||||
line_buffer = []
|
||||
@@ -283,7 +283,7 @@ def get_wrapped_help_text(
|
||||
return wrapped_help
|
||||
|
||||
|
||||
def wrap_text(text: str, wrap_width: int) -> list[str]:
|
||||
def wrap_text(text: str, wrap_width: int) -> List[str]:
|
||||
"""Wraps text while preserving spaces and breaking long words."""
|
||||
words = re.findall(r"\S+|\s+", text) # Capture words and spaces separately
|
||||
wrapped_lines = []
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
from typing import Any
|
||||
from typing import Any, Union, List, Dict
|
||||
|
||||
|
||||
class MenuState:
|
||||
def __init__(self):
|
||||
self.menu_index: list[int] = [] # Row we left the previous menus
|
||||
self.start_index: list[int] = [0] # Row to start the menu if it doesn't all fit
|
||||
self.menu_index: List[int] = [] # Row we left the previous menus
|
||||
self.start_index: List[int] = [0] # Row to start the menu if it doesn't all fit
|
||||
self.selected_index: int = 0 # Selected Row
|
||||
self.current_menu: dict[str, Any] | list[Any] | str | int = {} # Contents of the current menu
|
||||
self.menu_path: list[str] = [] # Menu Path
|
||||
self.current_menu: Union[Dict[str, Any], List[Any], str, int] = {} # Contents of the current menu
|
||||
self.menu_path: List[str] = [] # Menu Path
|
||||
self.show_save_option: bool = False # Display 'Save'
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import os
|
||||
import json
|
||||
import curses
|
||||
from typing import Any
|
||||
from typing import Any, List, Dict
|
||||
|
||||
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
|
||||
from contact.ui.default_config import format_json_single_line_arrays, loaded_config
|
||||
@@ -14,7 +14,7 @@ max_help_lines = 6
|
||||
save_option = "Save Changes"
|
||||
|
||||
|
||||
def edit_color_pair(key: str, current_value: list[str]) -> list[str]:
|
||||
def edit_color_pair(key: str, current_value: List[str]) -> List[str]:
|
||||
"""
|
||||
Allows the user to select a foreground and background color for a key.
|
||||
"""
|
||||
@@ -101,7 +101,7 @@ def edit_value(key: str, current_value: str) -> str:
|
||||
return user_input if user_input else current_value
|
||||
|
||||
|
||||
def display_menu(menu_state: Any) -> tuple[Any, Any, list[str]]:
|
||||
def display_menu(menu_state: Any) -> tuple[Any, Any, List[str]]:
|
||||
"""
|
||||
Render the configuration menu with a Save button directly added to the window.
|
||||
"""
|
||||
@@ -319,7 +319,7 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
|
||||
break
|
||||
|
||||
|
||||
def save_json(file_path: str, data: dict[str, Any]) -> None:
|
||||
def save_json(file_path: str, data: Dict[str, Any]) -> None:
|
||||
formatted_json = format_json_single_line_arrays(data)
|
||||
with open(file_path, "w", encoding="utf-8") as f:
|
||||
f.write(formatted_json)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import yaml
|
||||
import logging
|
||||
from typing import List
|
||||
from google.protobuf.json_format import MessageToDict
|
||||
from meshtastic import mt_config
|
||||
from meshtastic.util import camel_to_snake, snake_to_camel, fromStr
|
||||
@@ -20,9 +21,9 @@ def traverseConfig(config_root, config, interface_config) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def splitCompoundName(comp_name: str) -> list[str]:
|
||||
def splitCompoundName(comp_name: str) -> List[str]:
|
||||
"""Split compound (dot separated) preference name into parts"""
|
||||
name: list[str] = comp_name.split(".")
|
||||
name: List[str] = comp_name.split(".")
|
||||
if len(name) < 2:
|
||||
name[0] = comp_name
|
||||
name.append(comp_name)
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
from typing import Optional, Tuple, Dict, List
|
||||
import re
|
||||
|
||||
|
||||
def parse_ini_file(ini_file_path: str) -> tuple[dict[str, str], dict[str, str]]:
|
||||
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: str | None = None
|
||||
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:
|
||||
@@ -49,11 +50,11 @@ def parse_ini_file(ini_file_path: str) -> tuple[dict[str, str], dict[str, str]]:
|
||||
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."""
|
||||
path_replacements = {"Radio Settings": "config", "Module Settings": "module"}
|
||||
|
||||
transformed_path: list[str] = []
|
||||
transformed_path: List[str] = []
|
||||
for part in menu_path[1:]: # Skip 'Main Menu'
|
||||
# Apply fixed replacements
|
||||
part = path_replacements.get(part, part)
|
||||
|
||||
@@ -2,6 +2,7 @@ import sqlite3
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Optional, Union, Dict
|
||||
|
||||
from contact.utilities.utils import decimal_to_hex
|
||||
import contact.ui.default_config as config
|
||||
@@ -15,7 +16,7 @@ def get_table_name(channel: str) -> str:
|
||||
return quoted_table_name
|
||||
|
||||
|
||||
def save_message_to_db(channel: str, user_id: str, message_text: str) -> int | None:
|
||||
def save_message_to_db(channel: str, user_id: str, message_text: str) -> Optional[int]:
|
||||
"""Save messages to the database, ensuring the table exists."""
|
||||
try:
|
||||
quoted_table_name = get_table_name(channel)
|
||||
@@ -178,7 +179,7 @@ def init_nodedb() -> None:
|
||||
logging.error(f"Unexpected error in init_nodedb: {e}")
|
||||
|
||||
|
||||
def maybe_store_nodeinfo_in_db(packet: dict[str, object]) -> None:
|
||||
def maybe_store_nodeinfo_in_db(packet: Dict[str, object]) -> None:
|
||||
"""Save nodeinfo unless that record is already there, updating if necessary."""
|
||||
try:
|
||||
user_id = packet["from"]
|
||||
@@ -198,14 +199,14 @@ def maybe_store_nodeinfo_in_db(packet: dict[str, object]) -> None:
|
||||
|
||||
|
||||
def update_node_info_in_db(
|
||||
user_id: int | str,
|
||||
long_name: str | None = None,
|
||||
short_name: str | None = None,
|
||||
hw_model: str | None = None,
|
||||
is_licensed: str | int | None = None,
|
||||
role: str | None = None,
|
||||
public_key: str | None = None,
|
||||
chat_archived: int | None = None,
|
||||
user_id: Union[int, str],
|
||||
long_name: Optional[str] = None,
|
||||
short_name: Optional[str] = None,
|
||||
hw_model: Optional[str] = None,
|
||||
is_licensed: Optional[Union[str, int]] = None,
|
||||
role: Optional[str] = None,
|
||||
public_key: Optional[str] = None,
|
||||
chat_archived: Optional[int] = None,
|
||||
) -> None:
|
||||
"""Update or insert node information into the database, preserving unchanged fields."""
|
||||
try:
|
||||
|
||||
@@ -2,7 +2,7 @@ import base64
|
||||
import binascii
|
||||
import curses
|
||||
import ipaddress
|
||||
from typing import Any, Optional
|
||||
from typing import Any, Optional, List
|
||||
|
||||
from contact.ui.colors import get_color
|
||||
from contact.ui.nav_utils import move_highlight, draw_arrows, wrap_text
|
||||
@@ -98,7 +98,7 @@ def get_text_input(prompt: str) -> Optional[str]:
|
||||
return user_input
|
||||
|
||||
|
||||
def get_admin_key_input(current_value: list[bytes]) -> Optional[list[str]]:
|
||||
def get_admin_key_input(current_value: List[bytes]) -> Optional[List[str]]:
|
||||
def to_base64(byte_strings):
|
||||
"""Convert byte values to Base64-encoded strings."""
|
||||
return [base64.b64encode(b).decode() for b in byte_strings]
|
||||
@@ -183,7 +183,7 @@ def get_admin_key_input(current_value: list[bytes]) -> Optional[list[str]]:
|
||||
pass # Ignore invalid character inputs
|
||||
|
||||
|
||||
def get_repeated_input(current_value: list[str]) -> Optional[str]:
|
||||
def get_repeated_input(current_value: List[str]) -> Optional[str]:
|
||||
height = 9
|
||||
width = 80
|
||||
start_y = (curses.LINES - height) // 2
|
||||
@@ -309,7 +309,7 @@ def get_fixed32_input(current_value: int) -> int:
|
||||
pass # Ignore invalid inputs
|
||||
|
||||
|
||||
def get_list_input(prompt: str, current_option: Optional[str], list_options: list[str]) -> Optional[str]:
|
||||
def get_list_input(prompt: str, current_option: Optional[str], list_options: List[str]) -> Optional[str]:
|
||||
"""
|
||||
Displays a scrollable list of list_options for the user to choose from.
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user