forked from iarv/contact
Compare commits
12 Commits
dialog-scr
...
1.4.10
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f11f7bb9e0 | ||
|
|
ecd2d2d692 | ||
|
|
bdae90ecca | ||
|
|
56637f806b | ||
|
|
c6abedec75 | ||
|
|
6b18809215 | ||
|
|
b048fe2480 | ||
|
|
600fc61ed7 | ||
|
|
fbf5ff6bd3 | ||
|
|
faab1e961f | ||
|
|
255db3aa3c | ||
|
|
ad77eba0d6 |
@@ -42,7 +42,7 @@ For smaller displays you may wish to enable `single_pane_mode`:
|
||||
- `↑→↓←` = Navigate around the UI.
|
||||
- `F1/F2/F3` = Jump to Channel/Messages/Nodes
|
||||
- `ENTER` = Send a message typed in the Input Window, or with the Node List highlighted, select a node to DM
|
||||
- `` `` `or F12` = Open the Settings dialogue
|
||||
- `` ` `` or `F12` = Open the Settings dialogue
|
||||
- `CTRL` + `p` = Hide/show a log of raw received packets.
|
||||
- `CTRL` + `t` or `F4` = With the Node List highlighted, send a traceroute to the selected node
|
||||
- `F5` = Display a node's info
|
||||
|
||||
@@ -129,8 +129,10 @@ def start() -> None:
|
||||
|
||||
try:
|
||||
curses.wrapper(main)
|
||||
interface_state.interface.close()
|
||||
except KeyboardInterrupt:
|
||||
logging.info("User exited with Ctrl+C")
|
||||
interface_state.interface.close()
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
logging.critical("Fatal error", exc_info=True)
|
||||
|
||||
@@ -2,7 +2,46 @@ import logging
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import time
|
||||
import subprocess
|
||||
import threading
|
||||
# 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_lock = threading.Lock()
|
||||
_last_sound_request = 0.0
|
||||
|
||||
|
||||
def schedule_notification_sound(delay: float = _SOUND_DEBOUNCE_SECONDS) -> None:
|
||||
"""Schedule a notification sound after a short quiet period.
|
||||
|
||||
If more messages arrive before the delay elapses, the timer is reset.
|
||||
This prevents playing a sound for each message when a backlog flushes.
|
||||
"""
|
||||
global _sound_timer, _last_sound_request
|
||||
|
||||
now = time.monotonic()
|
||||
with _sound_timer_lock:
|
||||
_last_sound_request = now
|
||||
|
||||
# Cancel any previously scheduled sound.
|
||||
if _sound_timer is not None:
|
||||
try:
|
||||
_sound_timer.cancel()
|
||||
except Exception:
|
||||
pass
|
||||
_sound_timer = None
|
||||
|
||||
def _fire(expected_request_time: float) -> None:
|
||||
# Only play if nothing newer has been scheduled.
|
||||
with _sound_timer_lock:
|
||||
if expected_request_time != _last_sound_request:
|
||||
return
|
||||
play_sound()
|
||||
|
||||
_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 (
|
||||
@@ -108,7 +147,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
|
||||
|
||||
if config.notification_sound == "True":
|
||||
play_sound()
|
||||
schedule_notification_sound()
|
||||
|
||||
message_bytes = packet["decoded"]["payload"]
|
||||
message_string = message_bytes.decode("utf-8")
|
||||
|
||||
@@ -453,80 +453,85 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
|
||||
message_parts = []
|
||||
|
||||
message_parts.append("**📋 Basic Information:**")
|
||||
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')}")
|
||||
|
||||
role = f"{node.get('user', {}).get('role', 'Unknown')}"
|
||||
message_parts.append(f"• Role: {role}")
|
||||
|
||||
pk = f"{node.get('user', {}).get('publicKey')}"
|
||||
message_parts.append(f"Public key: {pk}")
|
||||
|
||||
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
|
||||
if "position" in node:
|
||||
pos = node["position"]
|
||||
if pos.get("latitude") and pos.get("longitude"):
|
||||
message_parts.append(f"• Position: {pos['latitude']:.4f}, {pos['longitude']:.4f}")
|
||||
if pos.get("altitude"):
|
||||
message_parts.append(f"• Altitude: {pos['altitude']}m")
|
||||
message_parts.append(f"https://maps.google.com/?q={pos['latitude']:.4f},{pos['longitude']:.4f}")
|
||||
|
||||
if any(key in node for key in ["snr", "hopsAway", "lastHeard"]):
|
||||
message_parts.append("\n**🌐 Network Metrics:**")
|
||||
|
||||
if "snr" in node:
|
||||
snr = node["snr"]
|
||||
snr_status = (
|
||||
"🟢 Excellent"
|
||||
if snr > 10
|
||||
else (
|
||||
"🟡 Good"
|
||||
if snr > 3
|
||||
else "🟠 Fair" if snr > -10 else "🔴 Poor" if snr > -20 else "💀 Very Poor"
|
||||
)
|
||||
)
|
||||
message_parts.append(f"• SNR: {snr}dB {snr_status}")
|
||||
|
||||
if "hopsAway" in node:
|
||||
hops = node["hopsAway"]
|
||||
hop_emoji = "📡" if hops == 0 else "🔄" if hops == 1 else "⏩"
|
||||
message_parts.append(f"• Hops away: {hop_emoji} {hops}")
|
||||
|
||||
if "lastHeard" in node and node["lastHeard"]:
|
||||
message_parts.append(f"• Last heard: 🕐 {get_time_ago(node['lastHeard'])}")
|
||||
|
||||
if node.get("deviceMetrics"):
|
||||
metrics = node["deviceMetrics"]
|
||||
message_parts.append("\n**📊 Device Metrics:**")
|
||||
|
||||
if "batteryLevel" in metrics:
|
||||
battery = metrics["batteryLevel"]
|
||||
battery_emoji = "🟢" if battery > 50 else "🟡" if battery > 20 else "🔴"
|
||||
voltage_info = f" ({metrics['voltage']}v)" if "voltage" in metrics else ""
|
||||
message_parts.append(f"• Battery: {battery_emoji} {battery}%{voltage_info}")
|
||||
|
||||
if "uptimeSeconds" in metrics:
|
||||
message_parts.append(f"• Uptime: ⏱️ {get_readable_duration(metrics['uptimeSeconds'])}")
|
||||
|
||||
if "channelUtilization" in metrics:
|
||||
util = metrics["channelUtilization"]
|
||||
util_emoji = "🔴" if util > 80 else "🟡" if util > 50 else "🟢"
|
||||
message_parts.append(f"• Channel utilization: {util_emoji} {util:.2f}%")
|
||||
|
||||
if "airUtilTx" in metrics:
|
||||
air_util = metrics["airUtilTx"]
|
||||
air_emoji = "🔴" if air_util > 80 else "🟡" if air_util > 50 else "🟢"
|
||||
message_parts.append(f"• Air utilization TX: {air_emoji} {air_util:.2f}%")
|
||||
|
||||
message = "\n".join(message_parts)
|
||||
|
||||
contact.ui.dialog.dialog(f"📡 Node Details: {node.get('user', {}).get('shortName', 'Unknown')}", message)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
message_parts = []
|
||||
|
||||
message_parts.append("**📋 Basic Information:**")
|
||||
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')}")
|
||||
|
||||
role = f"{node.get('user', {}).get('role', 'Unknown')}"
|
||||
message_parts.append(f"• Role: {role}")
|
||||
|
||||
pk = f"{node.get('user', {}).get('publicKey')}"
|
||||
message_parts.append(f"Public key: {pk}")
|
||||
|
||||
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
|
||||
if "position" in node:
|
||||
pos = node["position"]
|
||||
if pos.get("latitude") and pos.get("longitude"):
|
||||
message_parts.append(f"• Position: {pos['latitude']:.4f}, {pos['longitude']:.4f}")
|
||||
if pos.get("altitude"):
|
||||
message_parts.append(f"• Altitude: {pos['altitude']}m")
|
||||
message_parts.append(f"https://maps.google.com/?q={pos['latitude']:.4f},{pos['longitude']:.4f}")
|
||||
|
||||
if any(key in node for key in ["snr", "hopsAway", "lastHeard"]):
|
||||
message_parts.append("\n**🌐 Network Metrics:**")
|
||||
|
||||
if "snr" in node:
|
||||
snr = node["snr"]
|
||||
snr_status = (
|
||||
"🟢 Excellent"
|
||||
if snr > 10
|
||||
else "🟡 Good" if snr > 3 else "🟠 Fair" if snr > -10 else "🔴 Poor" if snr > -20 else "💀 Very Poor"
|
||||
)
|
||||
message_parts.append(f"• SNR: {snr}dB {snr_status}")
|
||||
|
||||
if "hopsAway" in node:
|
||||
hops = node["hopsAway"]
|
||||
hop_emoji = "📡" if hops == 0 else "🔄" if hops == 1 else "⏩"
|
||||
message_parts.append(f"• Hops away: {hop_emoji} {hops}")
|
||||
|
||||
if "lastHeard" in node and node["lastHeard"]:
|
||||
message_parts.append(f"• Last heard: 🕐 {get_time_ago(node['lastHeard'])}")
|
||||
|
||||
if node.get("deviceMetrics"):
|
||||
metrics = node["deviceMetrics"]
|
||||
message_parts.append("\n**📊 Device Metrics:**")
|
||||
|
||||
if "batteryLevel" in metrics:
|
||||
battery = metrics["batteryLevel"]
|
||||
battery_emoji = "🟢" if battery > 50 else "🟡" if battery > 20 else "🔴"
|
||||
voltage_info = f" ({metrics['voltage']}v)" if "voltage" in metrics else ""
|
||||
message_parts.append(f"• Battery: {battery_emoji} {battery}%{voltage_info}")
|
||||
|
||||
if "uptimeSeconds" in metrics:
|
||||
message_parts.append(f"• Uptime: ⏱️ {get_readable_duration(metrics['uptimeSeconds'])}")
|
||||
|
||||
if "channelUtilization" in metrics:
|
||||
util = metrics["channelUtilization"]
|
||||
util_emoji = "🔴" if util > 80 else "🟡" if util > 50 else "🟢"
|
||||
message_parts.append(f"• Channel utilization: {util_emoji} {util:.2f}%")
|
||||
|
||||
if "airUtilTx" in metrics:
|
||||
air_util = metrics["airUtilTx"]
|
||||
air_emoji = "🔴" if air_util > 80 else "🟡" if air_util > 50 else "🟢"
|
||||
message_parts.append(f"• Air utilization TX: {air_emoji} {air_util:.2f}%")
|
||||
|
||||
message = "\n".join(message_parts)
|
||||
|
||||
contact.ui.dialog.dialog(f"📡 Node Details: {node.get('user', {}).get('shortName', 'Unknown')}", message)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def handle_ctrl_t(stdscr: curses.window) -> None:
|
||||
"""Handle Ctrl + T key events to send a traceroute."""
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.4.5"
|
||||
version = "1.4.10"
|
||||
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"}
|
||||
]
|
||||
license = "GPL-3.0-only"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9,<3.14"
|
||||
requires-python = ">=3.9,<3.15"
|
||||
dependencies = [
|
||||
"meshtastic (>=2.6.0,<3.0.0)"
|
||||
"meshtastic (>=2.7.5,<3.0.0)"
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
|
||||
Reference in New Issue
Block a user