mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b5fd8d74c4 | ||
|
|
c383091a00 | ||
|
|
cc37f9a66b | ||
|
|
41ea441e32 | ||
|
|
58fb82fb1b | ||
|
|
dcd39c231f | ||
|
|
87bc876c3e | ||
|
|
10fc78c869 | ||
|
|
9fa66ac80f | ||
|
|
974a4af7f4 | ||
|
|
9026c56ebf | ||
|
|
26ca9599de | ||
|
|
44b2a3abee | ||
|
|
a26804b8b6 | ||
|
|
b225d5fe51 | ||
|
|
ea33b78af0 |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -8,4 +8,5 @@ client.log
|
||||
settings.log
|
||||
config.json
|
||||
default_config.log
|
||||
dist/
|
||||
dist/
|
||||
.vscode/launch.json
|
||||
|
||||
11
.vscode/launch.json
vendored
11
.vscode/launch.json
vendored
@@ -1,13 +1,22 @@
|
||||
{
|
||||
"version": "0.1.0",
|
||||
"configurations": [
|
||||
|
||||
{
|
||||
"name": "Python Debugger: Current File",
|
||||
"name": "Python Debugger: main",
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"cwd": "${workspaceFolder}",
|
||||
"module": "contact.__main__",
|
||||
"args": []
|
||||
},
|
||||
{
|
||||
"name": "Python Debugger: tcp",
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"cwd": "${workspaceFolder}",
|
||||
"module": "contact.__main__",
|
||||
"args": ["--host","192.168.86.69"]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -42,7 +42,9 @@ For smaller displays you may wish to enable `single_pane_mode`:
|
||||
- `ENTER` = Send a message typed in the Input Window, or with the Node List highlighted, select a node to DM
|
||||
- `` ` `` = Open the Settings dialogue
|
||||
- `CTRL` + `p` = Hide/show a log of raw received packets.
|
||||
- `CTRL` + `t` = With the Node List highlighted, send a traceroute to the selected node
|
||||
- `CTRL` + `t` = With the Node List highlighted, send a traceroute to the selected node
|
||||
- `CTRL` + `f` = With the Node List highlighted, favorite the selected node
|
||||
- `CTRL` + `g` = With the Node List highlighted, ignore the selected node
|
||||
- `CTRL` + `d` = With the Channel List hightlighted, archive a chat to reduce UI clutter. Messages will be saved in the db and repopulate if you send or receive a DM from this user.
|
||||
- `CTRL` + `d` = With the Note List highlghted, remove a node from your nodedb.
|
||||
- `ESC` = Exit out of the Settings Dialogue, or Quit the application if settings are not displayed.
|
||||
|
||||
@@ -101,6 +101,11 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
maybe_store_nodeinfo_in_db(packet)
|
||||
|
||||
elif packet["decoded"]["portnum"] == "TEXT_MESSAGE_APP":
|
||||
hop_start = packet.get('hopStart', 0)
|
||||
hop_limit = packet.get('hopLimit', 0)
|
||||
|
||||
hops = hop_start - hop_limit
|
||||
|
||||
|
||||
if config.notification_sound == "True":
|
||||
play_sound()
|
||||
@@ -140,7 +145,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
message_from_id = packet["from"]
|
||||
message_from_string = get_name_from_database(message_from_id, type="short") + ":"
|
||||
|
||||
add_new_message(channel_id, f"{config.message_prefix} {message_from_string} ", message_string)
|
||||
add_new_message(channel_id, f"{config.message_prefix} [{hops}] {message_from_string} ", message_string)
|
||||
|
||||
if refresh_channels:
|
||||
draw_channel_list()
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import time
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
import google.protobuf.json_format
|
||||
@@ -49,7 +51,7 @@ def onAckNak(packet: Dict[str, Any]) -> None:
|
||||
ack_type = "Nak"
|
||||
|
||||
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
|
||||
config.sent_message_prefix + confirm_string + ": ",
|
||||
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
|
||||
message,
|
||||
)
|
||||
|
||||
|
||||
@@ -97,14 +97,14 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
pkt_h = max(1, int(height / 3))
|
||||
|
||||
if firstrun:
|
||||
entry_win = curses.newwin(entry_height, width, 0, 0)
|
||||
entry_win = curses.newwin(entry_height, width, height - entry_height, 0)
|
||||
|
||||
channel_win = curses.newwin(content_h, channel_width, entry_height, 0)
|
||||
messages_win = curses.newwin(content_h, messages_width, entry_height, channel_width)
|
||||
nodes_win = curses.newwin(content_h, nodes_width, entry_height, channel_width + messages_width)
|
||||
channel_win = curses.newwin(content_h, channel_width, 0, 0)
|
||||
messages_win = curses.newwin(content_h, messages_width, 0, channel_width)
|
||||
nodes_win = curses.newwin(content_h, nodes_width, 0, channel_width + messages_width)
|
||||
|
||||
function_win = curses.newwin(function_height, width, height - function_height, 0)
|
||||
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - function_height, channel_width)
|
||||
function_win = curses.newwin(function_height, width, height - entry_height - function_height, 0)
|
||||
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height - function_height, channel_width)
|
||||
|
||||
# Will be resized to what we need when drawn
|
||||
messages_pad = curses.newpad(1, 1)
|
||||
@@ -127,21 +127,23 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
|
||||
for win in [entry_win, channel_win, messages_win, nodes_win, function_win, packetlog_win]:
|
||||
win.erase()
|
||||
|
||||
entry_win.resize(3, width)
|
||||
entry_win.resize(entry_height, width)
|
||||
entry_win.mvwin(height - entry_height, 0)
|
||||
|
||||
channel_win.resize(content_h, channel_width)
|
||||
channel_win.mvwin(0, 0)
|
||||
|
||||
messages_win.resize(content_h, messages_width)
|
||||
messages_win.mvwin(3, channel_width)
|
||||
messages_win.mvwin(0, channel_width)
|
||||
|
||||
nodes_win.resize(content_h, nodes_width)
|
||||
nodes_win.mvwin(entry_height, channel_width + messages_width)
|
||||
nodes_win.mvwin(0, channel_width + messages_width)
|
||||
|
||||
function_win.resize(3, width)
|
||||
function_win.mvwin(height - function_height, 0)
|
||||
function_win.resize(function_height, width)
|
||||
function_win.mvwin(height - entry_height - function_height, 0)
|
||||
|
||||
packetlog_win.resize(pkt_h, messages_width)
|
||||
packetlog_win.mvwin(height - pkt_h - function_height, channel_width)
|
||||
packetlog_win.mvwin(height - pkt_h - entry_height - function_height, channel_width)
|
||||
|
||||
# Draw window borders
|
||||
for win in [channel_win, entry_win, nodes_win, messages_win, function_win]:
|
||||
@@ -176,7 +178,7 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
handle_resize(stdscr, True)
|
||||
|
||||
while True:
|
||||
draw_text_field(entry_win, f"Input: {(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
|
||||
char = entry_win.get_wch()
|
||||
@@ -204,16 +206,22 @@ def main_ui(stdscr: curses.window) -> None:
|
||||
elif char == curses.KEY_LEFT or char == curses.KEY_RIGHT:
|
||||
handle_leftright(char)
|
||||
|
||||
elif char in (curses.KEY_F1, curses.KEY_F2, curses.KEY_F3):
|
||||
handle_function_keys(char)
|
||||
|
||||
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
|
||||
input_text = handle_enter(input_text)
|
||||
|
||||
elif char == chr(20): # Ctrl + t for Traceroute
|
||||
elif char in (curses.KEY_F4, chr(20)): # Ctrl + t and F4 for Traceroute
|
||||
handle_ctrl_t(stdscr)
|
||||
|
||||
elif char == curses.KEY_F5:
|
||||
handle_f5_key(stdscr)
|
||||
|
||||
elif char in (curses.KEY_BACKSPACE, chr(127)):
|
||||
input_text = handle_backspace(entry_win, input_text)
|
||||
|
||||
elif char == "`": # ` Launch the settings interface
|
||||
elif char in (curses.KEY_F12, "`"): # ` Launch the settings interface
|
||||
handle_backtick(stdscr)
|
||||
|
||||
elif char == chr(16): # Ctrl + P for Packet Log
|
||||
@@ -357,6 +365,51 @@ def handle_leftright(char: int) -> None:
|
||||
# Draw arrows last; force even in multi-pane to avoid flicker
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
def handle_function_keys(char: int) -> None:
|
||||
"""Switch windows using F1/F2/F3."""
|
||||
if char == curses.KEY_F1:
|
||||
target = 0
|
||||
elif char == curses.KEY_F2:
|
||||
target = 1
|
||||
elif char == curses.KEY_F3:
|
||||
target = 2
|
||||
else:
|
||||
return
|
||||
|
||||
old_window = ui_state.current_window
|
||||
|
||||
if target == old_window:
|
||||
return
|
||||
|
||||
ui_state.current_window = target
|
||||
handle_resize(root_win, False)
|
||||
|
||||
if old_window == 0:
|
||||
paint_frame(channel_win, selected=False)
|
||||
refresh_pad(0)
|
||||
elif old_window == 1:
|
||||
paint_frame(messages_win, selected=False)
|
||||
refresh_pad(1)
|
||||
elif old_window == 2:
|
||||
draw_function_win()
|
||||
paint_frame(nodes_win, selected=False)
|
||||
refresh_pad(2)
|
||||
|
||||
if not ui_state.single_pane_mode:
|
||||
draw_window_arrows(old_window)
|
||||
|
||||
if ui_state.current_window == 0:
|
||||
paint_frame(channel_win, selected=True)
|
||||
refresh_pad(0)
|
||||
elif ui_state.current_window == 1:
|
||||
paint_frame(messages_win, selected=True)
|
||||
refresh_pad(1)
|
||||
elif ui_state.current_window == 2:
|
||||
draw_function_win()
|
||||
paint_frame(nodes_win, selected=True)
|
||||
refresh_pad(2)
|
||||
|
||||
draw_window_arrows(ui_state.current_window)
|
||||
|
||||
def handle_enter(input_text: str) -> str:
|
||||
"""Handle Enter key events to send messages or select channels."""
|
||||
@@ -401,14 +454,104 @@ def handle_enter(input_text: str) -> str:
|
||||
return ""
|
||||
return input_text
|
||||
|
||||
def handle_f5_key(stdscr: curses.window) -> None:
|
||||
node = None
|
||||
try:
|
||||
node = interface_state.interface.nodesByNum[ui_state.node_list[ui_state.selected_node]]
|
||||
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."""
|
||||
now = time.monotonic()
|
||||
cooldown = 30.0
|
||||
remaining = cooldown - (now - ui_state.last_traceroute_time)
|
||||
|
||||
if remaining > 0:
|
||||
curses.curs_set(0) # Hide cursor
|
||||
contact.ui.dialog.dialog(
|
||||
"Traceroute Not Sent", f"Please wait {int(remaining)} seconds before sending another traceroute."
|
||||
)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
return
|
||||
|
||||
send_traceroute()
|
||||
ui_state.last_traceroute_time = now
|
||||
curses.curs_set(0) # Hide cursor
|
||||
contact.ui.dialog.dialog(
|
||||
f"Traceroute Sent To: {get_name_from_database(ui_state.node_list[ui_state.selected_node])}",
|
||||
"Results will appear in messages window.\nNote: Traceroute is limited to once every 30 seconds.",
|
||||
"Results will appear in messages window.",
|
||||
)
|
||||
curses.curs_set(1) # Show cursor again
|
||||
handle_resize(stdscr, False)
|
||||
@@ -634,7 +777,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
|
||||
for line in wrapped_lines:
|
||||
if prefix.startswith("--"):
|
||||
color = get_color("timestamps")
|
||||
elif prefix.startswith(config.sent_message_prefix):
|
||||
elif prefix.find(config.sent_message_prefix) != -1:
|
||||
color = get_color("tx_messages")
|
||||
else:
|
||||
color = get_color("rx_messages")
|
||||
@@ -684,9 +827,21 @@ def draw_node_list() -> None:
|
||||
for i, node_num in enumerate(ui_state.node_list):
|
||||
node = interface_state.interface.nodesByNum[node_num]
|
||||
secure = "user" in node and "publicKey" in node["user"] and node["user"]["publicKey"]
|
||||
node_str = f"{'🔐' if secure else '🔓'} {get_name_from_database(node_num, 'long')}".ljust(box_width - 2)[
|
||||
: box_width - 2
|
||||
]
|
||||
status_icon = '🔐' if secure else '🔓'
|
||||
node_name = get_name_from_database(node_num, 'long')
|
||||
user_name = node['user']['shortName']
|
||||
|
||||
uptime_str = ""
|
||||
if "deviceMetrics" in node and "uptimeSeconds" in node["deviceMetrics"]:
|
||||
uptime_str = f" / Up: {get_readable_duration(node['deviceMetrics']['uptimeSeconds'])}"
|
||||
|
||||
last_heard_str = f" ■ {get_time_ago(node['lastHeard'])}" if node.get("lastHeard") else ""
|
||||
hops_str = f" ■ Hops: {node['hopsAway']}" if "hopsAway" 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
|
||||
node_str = f"{status_icon} {node_name}"
|
||||
node_str = node_str.ljust(box_width - 4)[:box_width - 2]
|
||||
color = "node_list"
|
||||
if "isFavorite" in node and node["isFavorite"]:
|
||||
color = "node_favorite"
|
||||
@@ -972,11 +1127,13 @@ def draw_help() -> None:
|
||||
"""Draw the help text in the function window."""
|
||||
cmds = [
|
||||
"↑→↓← = Select",
|
||||
" F1/F2/F3 - Select windows",
|
||||
" ENTER = Send",
|
||||
" ` = Settings",
|
||||
" \"`\"/F12 = Settings",
|
||||
" ESC = Quit",
|
||||
" ^P = Packet Log",
|
||||
" ^t = Traceroute",
|
||||
" ^t/F4 = Traceroute",
|
||||
" F5 = Full node info",
|
||||
" ^d = Archive Chat",
|
||||
" ^f = Favorite",
|
||||
" ^g = Ignore",
|
||||
@@ -1016,6 +1173,7 @@ def refresh_pad(window: int) -> None:
|
||||
pad = nodes_pad
|
||||
box = nodes_win
|
||||
lines = box.getmaxyx()[0] - 2
|
||||
box.addstr(0, 2, (f"Nodes: {len(ui_state.node_list)}"), curses.A_BOLD)
|
||||
selected_item = ui_state.selected_node
|
||||
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ def draw_splash(stdscr: object) -> None:
|
||||
stdscr.addstr(start_y + 1, start_x - 1, message_2, get_color("splash_logo", bold=True))
|
||||
stdscr.addstr(start_y + 2, start_x - 2, message_3, get_color("splash_logo", bold=True))
|
||||
stdscr.addstr(start_y + 4, start_x2, message_4, get_color("splash_text"))
|
||||
stdscr.move(start_y + 5, start_x2)
|
||||
|
||||
stdscr.attrset(get_color("window_frame"))
|
||||
stdscr.box()
|
||||
|
||||
@@ -26,6 +26,7 @@ class ChatUIState:
|
||||
selected_node: int = 0
|
||||
current_window: int = 0
|
||||
last_sent_time: float = 0.0
|
||||
last_traceroute_time: float = 0.0
|
||||
|
||||
selected_index: int = 0
|
||||
start_index: List[int] = field(default_factory=lambda: [0, 0, 0])
|
||||
|
||||
@@ -136,13 +136,18 @@ def load_messages_from_db() -> None:
|
||||
elif ack_type == "Nak":
|
||||
ack_str = config.nak_str
|
||||
|
||||
ts_str = datetime.fromtimestamp(timestamp).strftime("[%H:%M:%S]")
|
||||
|
||||
if user_id == str(interface_state.myNodeNum):
|
||||
sanitized_message = message.replace("\x00", "")
|
||||
formatted_message = (f"{config.sent_message_prefix}{ack_str}: ", sanitized_message)
|
||||
formatted_message = (
|
||||
f"{ts_str} {config.sent_message_prefix}{ack_str}: ",
|
||||
sanitized_message,
|
||||
)
|
||||
else:
|
||||
sanitized_message = message.replace("\x00", "")
|
||||
formatted_message = (
|
||||
f"{config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
|
||||
f"{ts_str} {config.message_prefix} {get_name_from_database(int(user_id), 'short')}: ",
|
||||
sanitized_message,
|
||||
)
|
||||
|
||||
|
||||
@@ -167,7 +167,8 @@ def add_new_message(channel_id, prefix, message):
|
||||
ui_state.all_messages[channel_id].append((f"-- {current_hour} --", ""))
|
||||
|
||||
# Add the message
|
||||
ui_state.all_messages[channel_id].append((prefix, message))
|
||||
ts_str = time.strftime("[%H:%M:%S] ")
|
||||
ui_state.all_messages[channel_id].append((f"{ts_str}{prefix}", message))
|
||||
|
||||
|
||||
def parse_protobuf(packet: dict) -> Union[str, dict]:
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.4.0"
|
||||
version = "1.4.4"
|
||||
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