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

Move binary process to reader for consistent eventing

Browse source
This commit is contained in:
Alex Wolden 2025-08-29 11:54:21 -07:00
parent 28957a4b60
commit 9aeffb41a1
5 changed files with 257 additions and 165 deletions
Download patch file Download diff file Expand all files Collapse all files

43
examples/serial_repeater_telemetry.py
View file

@ -9,22 +9,20 @@ from meshcore.events import EventType
async def main():
# Parse command line arguments
parser = argparse.ArgumentParser(description='Get status from a repeater via serial connection')
# parser.add_argument('-p', '--port', required=True, help='Serial port')
# parser.add_argument('-b', '--baudrate', type=int, default=115200, help='Baud rate')
parser.add_argument('-p', '--port', required=True, help='Serial port')
parser.add_argument('-b', '--baudrate', type=int, default=115200, help='Baud rate')
parser.add_argument('-r', '--repeater', required=True, help='Repeater name')
parser.add_argument('-pw', '--password', required=True, help='Password for login')
# parser.add_argument('-t', '--timeout', type=float, default=10.0, help='Timeout for responses in seconds')
parser.add_argument('-t', '--timeout', type=float, default=10.0, help='Timeout for responses in seconds')
args = parser.parse_args()
# Connect to the device
mc = await MeshCore.create_ble("lora-py-tester")
534463
mc = await MeshCore.create_serial(args.port, args.baudrate, debug=True)
try:
# Get contacts
result = await mc.commands.get_contacts()
print(result)
print(mc._contacts)
repeater = mc.get_contact_by_key_prefix(args.repeater)
await mc.ensure_contacts()
repeater = mc.get_contact_by_name(args.repeater)
if repeater is None:
print(f"Repeater '{args.repeater}' not found in contacts.")
@ -37,25 +35,14 @@ async def main():
if login_event.type != EventType.ERROR:
print("Login successful")
# Continuously poll for telemetry every 60 seconds
print("Starting continuous telemetry polling every 60 seconds...")
while True:
try:
# Send status request
print("Sending status request...")
await mc.commands.send_telemetry_req(repeater)
# Wait for status response
telemetry_event = await mc.wait_for_event(EventType.TELEMETRY_RESPONSE, timeout=10)
print(telemetry_event)
# Wait 60 seconds before next poll
await asyncio.sleep(60)
except Exception as e:
print(f"Error during telemetry poll: {e}")
# Wait before retrying
await asyncio.sleep(60)
# Send status request
print("Sending status request...")
await mc.commands.send_telemetry_req(repeater)
# Wait for status response
telemetry_event = await mc.wait_for_event(EventType.TELEMETRY_RESPONSE, timeout=args.timeout)
print(telemetry_event.payload["lpp"])
else:
print("Login failed or timed out")