-
Notifications
You must be signed in to change notification settings - Fork 15
Expand file tree
/
Copy pathtest-rssi-snr.py
More file actions
58 lines (46 loc) · 1.67 KB
/
test-rssi-snr.py
File metadata and controls
58 lines (46 loc) · 1.67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import meshtastic.serial_interface, meshtastic.tcp_interface
from meshtastic.protobuf import telemetry_pb2, portnums_pb2
from pubsub import pub
import time
# dest = 147281321 ## DUMMY RAK
dest = 963936390 ## MESHSTICK
want_response = True
channel = 0
interface = meshtastic.serial_interface.SerialInterface()
interface = meshtastic.tcp_interface.TCPInterface(hostname="192.168.86.39")
telemetry_data = telemetry_pb2.Telemetry()
telemetry_data.device_metrics.battery_level = 69
response_received = False
last_request_time = None
def signalReport():
global last_request_time
global response_received
# print("Requesting data...")
response_received = False
last_request_time = time.time() # Update last request time
interface.sendData(
telemetry_data,
destinationId=dest,
portNum=portnums_pb2.PortNum.TELEMETRY_APP,
wantResponse=want_response
)
def onReceive(packet, interface):
global response_received
if packet['from'] == dest:
print(f"SNR: {packet['rxSnr']} RSSI: {packet['rxRssi']}")
response_received = True
pub.subscribe(onReceive, 'meshtastic.receive')
signalReport() # Initial request
try:
while True:
time.sleep(0.1) # Short sleep to allow quick response check
if response_received:
signalReport() # Send another request after receiving a response
elif time.time() - last_request_time > 15:
print("No response received for 10 seconds, retrying...")
signalReport() # Retry if no response received for 10 seconds
except KeyboardInterrupt:
print("Exiting program...")
finally:
interface.close()
print("Interface closed. Goodbye!")