meshcore-dev/meshcore-cli
1
0
Fork 0
mirror of https://github.com/meshcore-dev/meshcore-cli.git synced 2026-04-20 22:13:48 +00:00
Code Issues Projects Releases Packages Wiki Activity

sometimes we did not check if contacts existed

Browse source
This commit is contained in:
Florent 2026-01-03 11:16:26 +01:00
parent 23fb5c2fba
commit e34d826391
2 changed files with 174 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "meshcore-cli" name = "meshcore-cli"
version = "1.3.14" version = "1.3.15"
authors = [ authors = [
{ name="Florent de Lamotte", email="florent@frizoncorrea.fr" }, { name="Florent de Lamotte", email="florent@frizoncorrea.fr" },
] ]

289
src/meshcore_cli/meshcore_cli.py
View file

@ -32,7 +32,7 @@ import re
from meshcore import MeshCore, EventType, logger from meshcore import MeshCore, EventType, logger
# Version # Version
VERSION = "v1.3.14" VERSION = "v1.3.15"
# default ble address is stored in a config file # default ble address is stored in a config file
MCCLI_CONFIG_DIR = str(Path.home()) + "/.config/meshcore/" MCCLI_CONFIG_DIR = str(Path.home()) + "/.config/meshcore/"
@ -2679,38 +2679,56 @@ async def next_cmd(mc, cmds, json_output=False):
case "logout" : case "logout" :
argnum = 1 argnum = 1
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
res = await mc.commands.send_logout(contact) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "unknown contact"}))
print(f"Error while logout: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload))
else: else:
print("Logout ok") res = await mc.commands.send_logout(contact)
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error while logout: {res}")
elif json_output :
print(json.dumps(res.payload))
else:
print("Logout ok")
case "contact_timeout" : case "contact_timeout" :
argnum = 2 argnum = 2
contact = await get_contact_from_args(mc, cmds[1]) contact = await get_contact_from_args(mc, cmds[1])
contact["timeout"] = float(cmds[2]) if contact is None:
if json_output :
print(json.dumps({"error" : "unknown contact"}))
else:
print(f"Unknown contact {cmds[1]}")
else:
contact["timeout"] = float(cmds[2])
case "disc_path" | "dp" : case "disc_path" | "dp" :
argnum = 1 argnum = 1
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
res = await discover_path(mc, contact) if contact is None:
if res is None:
print(f"Error while discovering path")
else:
if json_output : if json_output :
print(json.dumps(res, indent=4)) print(json.dumps({"error" : "unknown contact"}))
else: else:
if "error" in res : print(f"Unknown contact {cmds[1]}")
print("Timeout while discovering path") else:
res = await discover_path(mc, contact)
if res is None:
print(f"Error while discovering path")
else:
if json_output :
print(json.dumps(res, indent=4))
else: else:
outp = res['out_path'] if "error" in res :
outp = outp if outp != "" else "direct" print("Timeout while discovering path")
inp = res['in_path'] else:
inp = inp if inp != "" else "direct" outp = res['out_path']
print(f"Path for {contact['adv_name']}: out {outp}, in {inp}") outp = outp if outp != "" else "direct"
inp = res['in_path']
inp = inp if inp != "" else "direct"
print(f"Path for {contact['adv_name']}: out {outp}, in {inp}")
case "node_discover"|"nd" : case "node_discover"|"nd" :
argnum = 1 argnum = 1
@ -2777,138 +2795,173 @@ async def next_cmd(mc, cmds, json_output=False):
argnum = 1 argnum = 1
await mc.ensure_contacts() await mc.ensure_contacts()
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
timeout = 0 if not "timeout" in contact else contact["timeout"] if contact is None:
res = await mc.commands.req_telemetry_sync(contact, timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting data") print(f"Unknown contact {cmds[1]}")
else : else:
print(json.dumps({ timeout = 0 if not "timeout" in contact else contact["timeout"]
"name": contact["adv_name"], res = await mc.commands.req_telemetry_sync(contact, timeout)
"pubkey_pre": contact["public_key"][0:16], if res is None :
"lpp": res, if json_output :
}, indent = 4)) print(json.dumps({"error" : "Getting data"}))
else:
print("Error getting data")
else :
print(json.dumps({
"name": contact["adv_name"],
"pubkey_pre": contact["public_key"][0:16],
"lpp": res,
}, indent = 4))
case "req_status"|"rs" : case "req_status"|"rs" :
argnum = 1 argnum = 1
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
timeout = 0 if not "timeout" in contact else contact["timeout"] if contact is None:
res = await mc.commands.req_status_sync(contact, timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting data") print(f"Unknown contact {cmds[1]}")
else : else:
print(json.dumps(res, indent=4)) timeout = 0 if not "timeout" in contact else contact["timeout"]
res = await mc.commands.req_status_sync(contact, timeout)
if res is None :
if json_output :
print(json.dumps({"error" : "Getting data"}))
else:
print("Error getting data")
else :
print(json.dumps(res, indent=4))
case "req_mma" | "rm": case "req_mma" | "rm":
argnum = 3 argnum = 3
await mc.ensure_contacts() await mc.ensure_contacts()
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
if cmds[2][-1] == "s": if contact is None:
from_secs = int(cmds[2][0:-1])
elif cmds[2][-1] == "m":
from_secs = int(cmds[2][0:-1]) * 60
elif cmds[2][-1] == "h":
from_secs = int(cmds[2][0:-1]) * 3600
else :
from_secs = int(cmds[2]) * 60 # same as tdeck
if cmds[3][-1] == "s":
to_secs = int(cmds[3][0:-1])
elif cmds[3][-1] == "m":
to_secs = int(cmds[3][0:-1]) * 60
elif cmds[3][-1] == "h":
to_secs = int(cmds[3][0:-1]) * 3600
else :
to_secs = int(cmds[3]) * 60
timeout = 0 if not "timeout" in contact else contact["timeout"]
res = await mc.commands.req_mma_sync(contact, from_secs, to_secs, timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting data") print(f"Unknown contact {cmds[1]}")
else : else:
print(json.dumps(res, indent=4)) if cmds[2][-1] == "s":
from_secs = int(cmds[2][0:-1])
elif cmds[2][-1] == "m":
from_secs = int(cmds[2][0:-1]) * 60
elif cmds[2][-1] == "h":
from_secs = int(cmds[2][0:-1]) * 3600
else :
from_secs = int(cmds[2]) * 60 # same as tdeck
if cmds[3][-1] == "s":
to_secs = int(cmds[3][0:-1])
elif cmds[3][-1] == "m":
to_secs = int(cmds[3][0:-1]) * 60
elif cmds[3][-1] == "h":
to_secs = int(cmds[3][0:-1]) * 3600
else :
to_secs = int(cmds[3]) * 60
timeout = 0 if not "timeout" in contact else contact["timeout"]
res = await mc.commands.req_mma_sync(contact, from_secs, to_secs, timeout)
if res is None :
if json_output :
print(json.dumps({"error" : "Getting data"}))
else:
print("Error getting data")
else :
print(json.dumps(res, indent=4))
case "req_acl" : case "req_acl" :
argnum = 1 argnum = 1
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
timeout = 0 if not "timeout" in contact else contact["timeout"] if contact is None:
res = await mc.commands.req_acl_sync(contact, timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting data") print(f"Unknown contact {cmds[1]}")
else : else:
if json_output: timeout = 0 if not "timeout" in contact else contact["timeout"]
print(json.dumps(res, indent=4)) res = await mc.commands.req_acl_sync(contact, timeout)
else: if res is None :
for e in res: if json_output :
name = e['key'] print(json.dumps({"error" : "Getting data"}))
ct = mc.get_contact_by_key_prefix(e['key']) else:
if ct is None: print("Error getting data")
if mc.self_info["public_key"].startswith(e['key']): else :
name = f"{'self':<20} [{e['key']}]" if json_output:
else: print(json.dumps(res, indent=4))
name = f"{ct['adv_name']:<20} [{e['key']}]" else:
print(f"{name:{' '}<35}: {e['perm']:02x}") for e in res:
name = e['key']
ct = mc.get_contact_by_key_prefix(e['key'])
if ct is None:
if mc.self_info["public_key"].startswith(e['key']):
name = f"{'self':<20} [{e['key']}]"
else:
name = f"{ct['adv_name']:<20} [{e['key']}]"
print(f"{name:{' '}<35}: {e['perm']:02x}")
case "req_neighbours"|"rn" : case "req_neighbours"|"rn" :
argnum = 1 argnum = 1
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
timeout = 0 if not "timeout" in contact else contact["timeout"] if contact is None:
res = await mc.commands.fetch_all_neighbours(contact, timeout=timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting data") print(f"Unknown contact {cmds[1]}")
else : else:
if json_output: timeout = 0 if not "timeout" in contact else contact["timeout"]
print(json.dumps(res, indent=4)) res = await mc.commands.fetch_all_neighbours(contact, timeout=timeout)
else: if res is None :
width = os.get_terminal_size().columns if json_output :
print(f"Got {res['results_count']} neighbours out of {res['neighbours_count']} from {contact['adv_name']}:") print(json.dumps({"error" : "Getting data"}))
for n in res['neighbours']: else:
ct = mc.get_contact_by_key_prefix(n["pubkey"]) print("Error getting data")
if ct and width > 60 : else :
name = f"[{n['pubkey'][0:8]}] {ct['adv_name']}" if json_output:
name = f"{name:30}" print(json.dumps(res, indent=4))
elif ct : else:
name = f"{ct['adv_name']}" width = os.get_terminal_size().columns
name = f"{name:20}" print(f"Got {res['results_count']} neighbours out of {res['neighbours_count']} from {contact['adv_name']}:")
else: for n in res['neighbours']:
name = f"[{n['pubkey']}]" ct = mc.get_contact_by_key_prefix(n["pubkey"])
if ct and width > 60 :
name = f"[{n['pubkey'][0:8]}] {ct['adv_name']}"
name = f"{name:30}"
elif ct :
name = f"{ct['adv_name']}"
name = f"{name:20}"
else:
name = f"[{n['pubkey']}]"
t_s = n['secs_ago'] t_s = n['secs_ago']
time_ago = f"{t_s}s" time_ago = f"{t_s}s"
if t_s / 86400 >= 1 : # result in days if t_s / 86400 >= 1 : # result in days
time_ago = f"{int(t_s/86400)}d ago{f' ({time_ago})' if width > 62 else ''}" time_ago = f"{int(t_s/86400)}d ago{f' ({time_ago})' if width > 62 else ''}"
elif t_s / 3600 >= 1 : # result in days elif t_s / 3600 >= 1 : # result in days
time_ago = f"{int(t_s/3600)}h ago{f' ({time_ago})' if width > 62 else ''}" time_ago = f"{int(t_s/3600)}h ago{f' ({time_ago})' if width > 62 else ''}"
elif t_s / 60 >= 1 : # result in min elif t_s / 60 >= 1 : # result in min
time_ago = f"{int(t_s/60)}m ago{f' ({time_ago})' if width > 62 else ''}" time_ago = f"{int(t_s/60)}m ago{f' ({time_ago})' if width > 62 else ''}"
print(f" {name} {time_ago}, {n['snr']}dB{' SNR' if width > 66 else ''}")
print(f" {name} {time_ago}, {n['snr']}dB{' SNR' if width > 66 else ''}")
case "req_binary" : case "req_binary" :
argnum = 2 argnum = 2
contact = await get_contact_from_arg(mc, cmds[1]) contact = await get_contact_from_arg(mc, cmds[1])
timeout = 0 if not "timeout" in contact else contact["timeout"] if contact is None:
res = await mc.commands.req_binary(contact, bytes.fromhex(cmds[2]), timeout)
if res is None :
if json_output : if json_output :
print(json.dumps({"error" : "Getting binary data"})) print(json.dumps({"error" : "unknown contact"}))
else: else:
print("Error getting binary data") print(f"Unknown contact {cmds[1]}")
else : else:
print(json.dumps(res)) timeout = 0 if not "timeout" in contact else contact["timeout"]
res = await mc.commands.req_binary(contact, bytes.fromhex(cmds[2]), timeout)
if res is None :
if json_output :
print(json.dumps({"error" : "Getting binary data"}))
else:
print("Error getting binary data")
else :
print(json.dumps(res))
case "contacts" | "list" | "lc": case "contacts" | "list" | "lc":
await mc.ensure_contacts(follow=True) await mc.ensure_contacts(follow=True)