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

verify contact

Browse source
This commit is contained in:
Florent de Lamotte 2025-04-17 22:51:37 +02:00
parent 0ee2fdb16a
commit 4324967b2b
1 changed files with 110 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

170
src/meshcore_cli/meshcore_cli.py
View file

@ -378,11 +378,11 @@ async def next_cmd(mc, cmds, json_output=False):
match cmds[1]: match cmds[1]:
case "help": case "help":
print("""Gets parameters from node print("""Gets parameters from node
name : node name name : node name
bat : battery level in mV bat : battery level in mV
coords : adv coordinates coords : adv coordinates
radio : radio parameters radio : radio parameters
tx : tx power""") tx : tx power""")
case "name": case "name":
if json_output : if json_output :
print(json.dumps(mc.self_info["name"])) print(json.dumps(mc.self_info["name"]))
@ -427,13 +427,19 @@ async def next_cmd(mc, cmds, json_output=False):
argnum = 2 argnum = 2
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.send_msg(contact, cmds[2]) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error sending message: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
res.payload["expected_ack"] = res.payload["expected_ack"].hex() else:
print(json.dumps(res.payload, indent=4)) res = await mc.commands.send_msg(contact, cmds[2])
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error sending message: {res}")
elif json_output :
res.payload["expected_ack"] = res.payload["expected_ack"].hex()
print(json.dumps(res.payload, indent=4))
case "chan"|"ch" : case "chan"|"ch" :
argnum = 2 argnum = 2
@ -457,28 +463,40 @@ async def next_cmd(mc, cmds, json_output=False):
argnum = 2 argnum = 2
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.send_cmd(contact, cmds[2]) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error sending cmd: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
res.payload["expected_ack"] = res.payload["expected_ack"].hex() else:
print(json.dumps(res.payload, indent=4)) res = await mc.commands.send_cmd(contact, cmds[2])
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error sending cmd: {res}")
elif json_output :
res.payload["expected_ack"] = res.payload["expected_ack"].hex()
print(json.dumps(res.payload, indent=4))
case "login" | "l" | "[[" : case "login" | "l" | "[[" :
argnum = 2 argnum = 2
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.send_login(contact, cmds[2]) if contact is None:
logger.debug(res)
if res.type == EventType.ERROR:
if json_output : if json_output :
print(json.dumps({"error" : "Error while login"})) print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
else: else:
print(f"Error while loging: {res}") print(f"Unknown contact {cmds[1]}")
elif json_output : else:
res.payload["expected_ack"] = res.payload["expected_ack"].hex() res = await mc.commands.send_login(contact, cmds[2])
print(json.dumps(res.payload, indent=4)) logger.debug(res)
if res.type == EventType.ERROR:
if json_output :
print(json.dumps({"error" : "Error while login"}))
else:
print(f"Error while loging: {res}")
elif json_output :
res.payload["expected_ack"] = res.payload["expected_ack"].hex()
print(json.dumps(res.payload, indent=4))
case "logout" : case "logout" :
argnum = 1 argnum = 1
@ -517,49 +535,73 @@ async def next_cmd(mc, cmds, json_output=False):
argnum = 2 argnum = 2
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.change_contact_path(contact, cmds[2]) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error setting path: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload, indent=4)) else:
await mc.commands.get_contacts() res = await mc.commands.change_contact_path(contact, cmds[2])
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error setting path: {res}")
elif json_output :
print(json.dumps(res.payload, indent=4))
await mc.commands.get_contacts()
case "reset_path" | "rp" : case "reset_path" | "rp" :
argnum = 1 argnum = 1
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.reset_path(contact) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error resetting path: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload, indent=4)) else:
await mc.commands.get_contacts() res = await mc.commands.reset_path(contact)
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error resetting path: {res}")
elif json_output :
print(json.dumps(res.payload, indent=4))
await mc.commands.get_contacts()
case "share_contact" | "sc": case "share_contact" | "sc":
argnum = 1 argnum = 1
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.share_contact(contact) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error while sharing contact: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload, indent=4)) else:
res = await mc.commands.share_contact(contact)
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error while sharing contact: {res}")
elif json_output :
print(json.dumps(res.payload, indent=4))
case "export_contact"|"ec": case "export_contact"|"ec":
argnum = 1 argnum = 1
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.export_contact(contact) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error exporting contact: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload, indent=4)) else:
else : res = await mc.commands.export_contact(contact)
print(res.payload) logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error exporting contact: {res}")
elif json_output :
print(json.dumps(res.payload, indent=4))
else :
print(res.payload)
case "card" : case "card" :
res = await mc.commands.export_contact() res = await mc.commands.export_contact()
@ -575,12 +617,18 @@ async def next_cmd(mc, cmds, json_output=False):
argnum = 1 argnum = 1
await mc.ensure_contacts() await mc.ensure_contacts()
contact = mc.get_contact_by_name(cmds[1]) contact = mc.get_contact_by_name(cmds[1])
res = await mc.commands.remove_contact(contact) if contact is None:
logger.debug(res) if json_output :
if res.type == EventType.ERROR: print(json.dumps({"error" : "contact unknown", "name" : cmds[1]}))
print(f"Error removing contact: {res}") else:
elif json_output : print(f"Unknown contact {cmds[1]}")
print(json.dumps(res.payload, indent=4)) else:
res = await mc.commands.remove_contact(contact)
logger.debug(res)
if res.type == EventType.ERROR:
print(f"Error removing contact: {res}")
elif json_output :
print(json.dumps(res.payload, indent=4))
case "recv" | "r" : case "recv" | "r" :
res = await mc.commands.get_msg() res = await mc.commands.get_msg()
@ -665,6 +713,8 @@ async def next_cmd(mc, cmds, json_output=False):
print("Timeout waiting ack") print("Timeout waiting ack")
elif json_output : elif json_output :
print(json.dumps(res.payload, indent=4)) print(json.dumps(res.payload, indent=4))
else :
print("Msg acked")
case "wait_login" | "wl" | "]]": case "wait_login" | "wl" | "]]":
res = await mc.wait_for_event(EventType.LOGIN_SUCCESS) res = await mc.wait_for_event(EventType.LOGIN_SUCCESS)