-
Notifications
You must be signed in to change notification settings - Fork 1
/
LoraLog.py
1811 lines (1617 loc) · 93.3 KB
/
LoraLog.py
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import time
from datetime import datetime
import sys
import asyncio
import gc
import psutil
import math
from unidecode import unidecode
from configparser import ConfigParser
import pickle
from html import unescape
from pygame import mixer
import threading
import copy
# import yaml
# Tkinter imports
from PIL import Image, ImageTk
import tkinter as tk
import customtkinter
from tkinter import Frame, LabelFrame, ttk
from tkintermapview2 import TkinterMapView
import textwrap
# Meshtastic imports
import base64
from pubsub import pub
import meshtastic.remote_hardware
import meshtastic.version
import meshtastic.tcp_interface
import meshtastic.serial_interface
try:
from meshtastic.protobuf import config_pb2
except ImportError:
from meshtastic import config_pb2
'''
Fix sub parts if they brake a main part install > pip install --upgrade setuptools <sub tool name>
Upgrade the Meshtastic Python Library > pip install --upgrade meshtastic
Build the build > pyinstaller --icon=mesh.ico -F --onefile --noconsole LoraLog.py
from pprint import pprint
pprint(vars(object))
'''
# Configure Error logging
import logging
logging.basicConfig(filename='LoraLog.log', level=logging.ERROR, format='%(asctime)s : %(message)s', datefmt='%m-%d %H:%M', filemode='w')
logging.error("Startin Up")
def has_pairs(lst):
return len(lst) != 0 and len(lst) % 2 == 0
config = ConfigParser()
config.read('config.ini')
telemetry_thread = None
position_thread = None
trace_thread = None
MapMarkers = {}
ok2Send = 0
isConnect = False
MyLora = ''
MyLoraText1 = ''
MyLoraText2 = ''
mylorachan = {}
tlast = int(time.time())
loop = None
pingcount = 0
def showLink(event):
idx= event.widget.tag_names("current")[1]
temp = type('temp', (object,), {})()
temp.data = idx
if idx in LoraDB:
click_command(temp)
else:
logging.error(f'Node {idx} not in DB')
# Function to insert colored text
def insert_colored_text(text_widget, text, color, center=False, tag=None):
global hr_img, MyLora
parent_frame = str(text_widget.winfo_parent())
if "frame5" not in parent_frame:
text_widget.configure(state="normal")
if color == '#d1d1d1': # and "frame3" not in parent_frame:
text_widget.image_create("end", image=hr_img)
text_widget.tag_configure(color, foreground=color)
if tag: # and tag != MyLora:
text_widget.tag_configure(tag, foreground=color, underline=False)
text_widget.insert(tk.END, text, (color, tag))
text_widget.tag_bind(tag, "<Button-1>", showLink)
else:
text_widget.insert(tk.END, text, color)
if center:
text_widget.tag_configure("center", justify='center')
text_widget.tag_add("center", "1.0", "end")
if "!frame5" not in parent_frame:
text_widget.see(tk.END)
text_widget.configure(state="disabled")
def add_message(text_widget, nodeid, mtext, msgtime, private=False, msend='all', ackn=True, bulk=False):
label = LoraDB[nodeid][1] + " (" + LoraDB[nodeid][2] + ")"
tcolor = "#00c983"
if nodeid == MyLora: tcolor = "#2bd5ff"
timestamp = datetime.fromtimestamp(msgtime).strftime("%Y-%m-%d %H:%M:%S")
text_widget.image_create("end", image=hr_img)
insert_colored_text(text_widget,'\n From ' + unescape(label),tcolor)
if private:
insert_colored_text(text_widget,' [Ch ' + private + ']', "#c9a500")
ptext = unescape(mtext).strip()
ptext = textwrap.fill(ptext, 87)
tcolor = "#d2d2d2"
if bulk == True:
tcolor = "#a1a1a1"
ptext = textwrap.indent(text=ptext, prefix=' ', predicate=lambda line: True)
insert_colored_text(text_widget, '\n' + ptext + '\n', tcolor)
insert_colored_text(text_widget,timestamp.rjust(89) + '\n', "#818181")
if bulk == False:
# We might have to html it so we dont get any ' and " in text that would break the the db
chat_log.append({'nodeID': nodeid, 'time': msgtime, 'private': private, 'send': msend, 'ackn': ackn, 'seen': False, 'text': str(mtext.encode('ascii', 'xmlcharrefreplace'), 'ascii')})
# url escape for save storage > str(text.encode('ascii', 'xmlcharrefreplace'), 'ascii')
# and back to normal > unescape(text_from)
# Do wee need to send ackn back to the sender after we seen the message ?
def get_messages():
sorted_data = chat_log[-10:] # for now retrieve only the last 10 messages
for entry in sorted_data:
add_message(text_box3, entry['nodeID'], unescape(entry['text']), entry['time'], private=entry['private'], msend=entry['send'], ackn=entry['ackn'], bulk=True)
#------------------------------------------------------------- Movment Tracker --------------------------------------------------------------------------
movement_log = [] # movement_log = [{'nodeID': '1', 'time': 1698163200, 'latitude': 10.0, 'longitude': 20.0, 'altitude': 1000}, ...]
metrics_log = [] # metrics_log = [{'nodeID': '1', 'time': 1698163200, 'battery': 100, 'voltage': 3.7, 'utilization': 0.0, 'airutiltx': 0.0}, ...]
environment_log = [] # environment = [{'nodeID': '1', 'time': 1698163200, 'temperature': 1.0, 'humidity': 20.0, 'pressure': 1010.0}, ...]
LoraDB = {} # LoraDB = {'nodeID': [timenow, ShortName, LongName, latitude, longitude, altitude, macaddr, hardware, timefirst, rightbarstats, mmqtt, snr, hops], ...}
chat_log = [] # chat_log = [{'nodeID': '1', 'time': 1698163200, 'private', True, 'send': 'nodeid or ch', 'ackn' : True, seen': False, 'text': 'Hello World!'}, ...]
# Load the databases
LoraDBPath = 'DataBase' + os.path.sep + 'LoraDB.pkl'
if os.path.exists(LoraDBPath):
with open(LoraDBPath, 'rb') as f:
LoraDB = pickle.load(f)
MoveDBPath = 'DataBase' + os.path.sep + 'MoveDB.pkl'
if os.path.exists(MoveDBPath):
with open(MoveDBPath, 'rb') as f:
movement_log = pickle.load(f)
MetricsPath = 'DataBase' + os.path.sep + 'MetricsDB.pkl'
if os.path.exists(MetricsPath):
with open(MetricsPath, 'rb') as f:
metrics_log = pickle.load(f)
EnviPath = 'DataBase' + os.path.sep + 'EnviDB.pkl'
if os.path.exists(EnviPath):
with open(EnviPath, 'rb') as f:
environment_log = pickle.load(f)
ChatPath = 'DataBase' + os.path.sep + 'ChatDB.pkl'
if os.path.exists(ChatPath):
with open(ChatPath, 'rb') as f:
chat_log = pickle.load(f)
if len(LoraDB) > 1:
logging.error("Loaded LoraDB with " + str(len(LoraDB)) + " entries")
def get_last_position(database, nodeID):
for entry in reversed(database):
if entry['nodeID'] == nodeID:
return entry
return None
def get_first_position(database, nodeID):
for entry in database:
if entry['nodeID'] == nodeID:
return entry
return None
def count_entries_for_node(database, nodeID):
return len([entry for entry in database if entry['nodeID'] == nodeID])
def get_data_for_node(database, nodeID):
data = [entry for entry in database if entry['nodeID'] == nodeID]
return data
def safedatabase():
global LoraDB, LoraDBPath, movement_log, MoveDBPath, metrics_log, MetricsPath, environment_log, EnviPath
if not os.path.exists('DataBase'):
os.makedirs('DataBase')
with open(LoraDBPath, 'wb') as f:
pickle.dump(LoraDB, f)
with open(MoveDBPath, 'wb') as f:
pickle.dump(movement_log, f)
with open(MetricsPath, 'wb') as f:
pickle.dump(metrics_log, f)
with open(EnviPath, 'wb') as f:
pickle.dump(environment_log, f)
with open(ChatPath, 'wb') as f:
pickle.dump(chat_log, f)
logging.error("Database saved!")
#----------------------------------------------------------- Meshtastic Lora Con ------------------------------------------------------------------------
meshtastic_client = None
try:
map_delete = int(config.get('meshtastic', 'map_delete_time')) * 60
map_oldnode = int(config.get('meshtastic', 'map_oldnode_time')) * 60
map_trail_age = int(config.get('meshtastic', 'map_trail_age')) * 60
metrics_age = int(config.get('meshtastic', 'metrics_age')) * 60
except Exception as e :
logging.error("Error loading databases: %s", str(e))
map_delete = 2700
map_oldnode = 86400
map_trail_age = 43200
metrics_age = 86400
mixer.init()
sound_cache = {}
def playsound(soundfile):
if soundfile not in sound_cache:
sound_cache[soundfile] = mixer.Sound(soundfile)
sound_cache[soundfile].play()
def value_to_graph(value, min_value=-19, max_value=1, graph_length=12):
value = max(min_value, min(max_value, value))
position = int((value - min_value) / (max_value - min_value) * (graph_length - 1))
position0 = int((0 - min_value) / (max_value - min_value) * (graph_length - 1))
graph = ['─'] * graph_length
graph[position0] = '┴'
graph[position] = '╥'
return '└' + ''.join(graph) + '┘'
def connect_meshtastic(force_connect=False):
global meshtastic_client, MyLora, movement_log, loop, isLora, isConnect
if meshtastic_client and not force_connect:
return meshtastic_client
# Initialize the event loop
try:
loop = asyncio.get_event_loop()
# loop.run_forever()
if loop.is_closed():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
meshtastic_client = None
# Initialize Meshtastic interface
retry_limit = 3
attempts = 1
successful = False
target_host = config.get('meshtastic', 'host')
comport = config.get('meshtastic', 'serial_port')
cnto = target_host
if config.get('meshtastic', 'interface') != 'tcp':
cnto = comport
logging.debug("Connecting to meshtastic on " + cnto + "...")
insert_colored_text(text_box1, " Connecting to meshtastic on " + cnto + "...\n", "#00c983")
while not successful and attempts <= retry_limit:
try:
if config.get('meshtastic', 'interface') == 'tcp':
meshtastic_client = meshtastic.tcp_interface.TCPInterface(hostname=target_host)
else:
meshtastic_client = meshtastic.serial_interface.SerialInterface(comport)
successful = True
except Exception as e:
attempts += 1
if attempts <= retry_limit:
logging.error("Connect re-try: ", str(e))
time.sleep(3)
else:
logging.error("Could not connect: s", str(e))
isLora = False
return None
isConnect = True
nodeInfo = meshtastic_client.getMyNodeInfo()
logging.debug("Connected to " + nodeInfo['user']['id'] + " > " + nodeInfo['user']['shortName'] + " / " + nodeInfo['user']['longName'] + " using a " + nodeInfo['user']['hwModel'])
insert_colored_text(text_box1, " Connected to " + nodeInfo['user']['id'] + " > " + nodeInfo['user']['shortName'] + " / " + nodeInfo['user']['longName'] + " using a " + nodeInfo['user']['hwModel'] + "\n", "#00c983")
MyLora = (nodeInfo['user']['id'])[1:]
pub.subscribe(on_meshtastic_message, "meshtastic.receive", loop=asyncio.get_event_loop())
pub.subscribe(on_meshtastic_connection, "meshtastic.connection.established")
pub.subscribe(on_lost_meshtastic_connection,"meshtastic.connection.lost")
print("MyLora: " + MyLora)
root.wm_title("Meshtastic Lora Logger - " + unescape(LoraDB[MyLora][1]))
logLora((nodeInfo['user']['id'])[1:], ['NODEINFO_APP', nodeInfo['user']['shortName'], nodeInfo['user']['longName'], nodeInfo['user']["macaddr"],nodeInfo['user']['hwModel']])
nodeInfo = meshtastic_client.getNode('^local')
# Lets get the Local Node's channels
lora_config = nodeInfo.localConfig.lora
modem_preset_enum = lora_config.modem_preset
modem_preset_string = config_pb2._CONFIG_LORACONFIG_MODEMPRESET.values_by_number[modem_preset_enum].name
channels = nodeInfo.channels
if channels:
for channel in channels:
psk_base64 = base64.b64encode(channel.settings.psk).decode('utf-8')
if channel.settings.name == '':
mylorachan[channel.index] = str(channel.index)
else:
mylorachan[channel.index] = unidecode(channel.settings.name)
if channel.index == 0 and mylorachan[channel.index] == '0':
mylorachan[channel.index] = modem_preset_string
if channel.index == 0:
insert_colored_text(text_box1, " Lora Chat Channel 0 = " + mylorachan[0] + " using Key " + psk_base64 + "\n", "#00c983")
padding_frame.config(text="Send a message to channel " + mylorachan[0])
updatesnodes()
return meshtastic_client
def req_meta():
global meshtastic_client, loop, ok2Send
try:
meshtastic_client.localNode.getMetadata()
except Exception as e:
logging.error("Error requesting metadata: %s", str(e))
finally:
print(f"Finished requesting metadata")
ok2Send = 0
def on_lost_meshtastic_connection(interface):
global root, loop, telemetry_thread, position_thread, trace_thread, isConnect
safedatabase()
isConnect = False
logging.error("Lost connection to Meshtastic Node.")
if telemetry_thread != None and telemetry_thread.is_alive():
telemetry_thread.join()
if position_thread != None and position_thread.is_alive():
position_thread.join()
if trace_thread != None and trace_thread.is_alive():
trace_thread.join()
pub.unsubscribe(on_meshtastic_message, "meshtastic.receive")
pub.unsubscribe(on_meshtastic_connection, "meshtastic.connection.established")
pub.unsubscribe(on_lost_meshtastic_connection, "meshtastic.connection.lost")
insert_colored_text(text_box1, "[" + time.strftime("%H:%M:%S", time.localtime()) + "]", "#d1d1d1")
insert_colored_text(text_box1, " Lost connection to node!", "#db6544")
root.meshtastic_interface = None
if isLora:
logging.error("Trying to re-connect...")
insert_colored_text(text_box1, ", Trying to re-connect...\n", "#db6544")
time.sleep(3)
root.meshtastic_interface = connect_meshtastic(force_connect=True)
def on_meshtastic_connection(interface, topic=pub.AUTO_TOPIC):
print("Connected to meshtastic")
def logLora(nodeID, info):
global LoraDB
tnow = int(time.time())
if nodeID in LoraDB:
LoraDB[nodeID][0] = tnow # time last seen
else:
LoraDB[nodeID] = [tnow, nodeID[-4:], '', -8.0, -8.0, 0, '', '', tnow, '0% 0.0v', '', '',-1, 0]
insert_colored_text(text_box1, "[" + time.strftime("%H:%M:%S", time.localtime()) + "]", "#d1d1d1")
insert_colored_text(text_box1, " New Node Logged [!" + nodeID + "]\n", "#e8643f", tag=nodeID)
if info[0] == 'NODEINFO_APP':
tmp = str(info[1].encode('ascii', 'xmlcharrefreplace'), 'ascii').replace("\n", "") # short name
if tmp != '':
LoraDB[nodeID][1] = tmp
else:
LoraDB[nodeID][1] = nodeID[-4:]
tmp = str(info[2].encode('ascii', 'xmlcharrefreplace'), 'ascii').replace("\n", "") # long name
if tmp != '':
LoraDB[nodeID][2] = tmp
else:
LoraDB[nodeID][2] = '!' + nodeID
LoraDB[nodeID][6] = info[3] # mac adress
LoraDB[nodeID][7] = info[4] # hardware
elif info[0] == 'POSITION_APP':
LoraDB[nodeID][3] = info[1] # latitude
LoraDB[nodeID][4] = info[2] # longitude
LoraDB[nodeID][5] = info[3] # altitude
def print_range(range_in_meters):
if range_in_meters < 1:
# Convert to centimeters
range_in_cm = range_in_meters * 100
return f"{range_in_cm:.0f}cm"
elif range_in_meters < 1000:
# Print in meters
return f"{range_in_meters:.0f}meter"
else:
# Convert to kilometers
range_in_km = range_in_meters / 1000
return f"{range_in_km:.0f}km"
def idToHex(nodeId):
in_hex = hex(nodeId)
if len(in_hex)%2: in_hex = in_hex.replace("0x","0x0") # Need account for leading zero, wish hex removes if it has one
return f"!{in_hex[2:]}"
def MapMarkerDelete(node_id):
global MapMarkers
if node_id in MapMarkers:
# Mheard
if MapMarkers[node_id][3] != None:
MapMarkers[node_id][3].delete()
MapMarkers[node_id][3] = None
# Move Trail
if MapMarkers[node_id][4] != None:
MapMarkers[node_id][4].delete()
MapMarkers[node_id][4] = None
# Check Trail
MapMarkers[node_id][5] = 0
# Range Circle
if len(MapMarkers[node_id]) == 8:
if MapMarkers[node_id][7] != None:
MapMarkers[node_id][7].delete()
MapMarkers[node_id][7] = None
MapMarkers[node_id].pop()
def on_meshtastic_message(packet, interface, loop=None):
# print(yaml.dump(packet))
global MyLora, MyLoraText1, MyLoraText2, LoraDB, MapMarkers, movement_log
if MyLora == '':
print('*** MyLora is empty ***\n')
return
ischat = False
tnow = int(time.time())
rectime = tnow
if 'rxTime' in packet:
rectime = packet['rxTime']
text_from = ''
if 'fromId' in packet and packet['fromId'] is not None:
text_from = packet.get('fromId', '')[1:]
if text_from == '':
text_from = idToHex(packet["from"])[1:]
fromraw = text_from
if "decoded" in packet:
data = packet["decoded"]
if text_from !='':
viaMqtt = False
text_msgs = ''
if text_from in LoraDB:
LoraDB[text_from][0] = tnow
if LoraDB[text_from][1] != '':
text_from = LoraDB[text_from][1] + " (" + LoraDB[text_from][2] + ")"
else:
LoraDB[text_from] = [tnow, fromraw[-4:], '', -8.0, -8.0, 0, '', '', tnow, '0% 0.0v', '', '', -1, 0]
insert_colored_text(text_box1, "[" + time.strftime("%H:%M:%S", time.localtime()) + "]", "#d1d1d1")
insert_colored_text(text_box1, " New Node Logged [!" + fromraw + "]\n", "#e8643f", tag=fromraw)
playsound('Data' + os.path.sep + 'NewNode.mp3')
if "viaMqtt" in packet:
LoraDB[fromraw][10] = ' via mqtt'
viaMqtt = True
else:
LoraDB[fromraw][10] = ''
LoraDB[fromraw][12] = -1
if "hopStart" in packet: LoraDB[fromraw][12] = packet['hopStart']
# Lets Work the Msgs
if data["portnum"] == "ADMIN_APP":
if "getDeviceMetadataResponse" in data["admin"]:
text_raws = f"Firmware version : {data['admin']['getDeviceMetadataResponse']['firmwareVersion']}"
else:
text_raws = 'Admin Data'
elif data["portnum"] == "TELEMETRY_APP":
text_raws = 'Node Telemetry'
telemetry = packet['decoded'].get('telemetry', {})
if telemetry:
device_metrics = telemetry.get('deviceMetrics', {})
if device_metrics:
LoraDB[fromraw][9] = ''
text_raws += '\n' + (' ' * 11) + 'Battery: ' + str(device_metrics.get('batteryLevel', 0)) + '% '
if device_metrics.get('batteryLevel', 0) < 101:
LoraDB[fromraw][9] = str(device_metrics.get('batteryLevel', 0)) + '% '
text_raws += 'Power: ' + str(round(device_metrics.get('voltage', 0.00),2)) + 'v '
LoraDB[fromraw][9] += str(round(device_metrics.get('voltage', 0.00),2)) + 'v'
text_raws += 'ChUtil: ' + str(round(device_metrics.get('channelUtilization', 0.00),2)) + '% '
text_raws += 'AirUtilTX (DutyCycle): ' + str(round(device_metrics.get('airUtilTx', 0.00),2)) + '%'
if len(LoraDB[fromraw]) < 14:
LoraDB[fromraw].append(0)
logging.error(f"Node {fromraw} has no uptime and a length of {len(LoraDB[fromraw])}")
LoraDB[fromraw][13] = device_metrics.get('uptimeSeconds', 0)
text_raws += '\n' + (' ' * 11) + uptimmehuman(fromraw)
# Need store uptimme somwhere !
if MyLora == fromraw:
MyLoraText1 = (' ChUtil').ljust(13) + str(round(device_metrics.get('channelUtilization', 0.00),2)).rjust(6) + '%\n' + (' AirUtilTX').ljust(13) + str(round(device_metrics.get('airUtilTx', 0.00),2)).rjust(6) + '%\n' + (' Power').ljust(13) + str(round(device_metrics.get('voltage', 0.00),2)).rjust(6) + 'v\n' + (' Battery').ljust(13) + str(device_metrics.get('batteryLevel', 0)).rjust(6) + '%\n'
if 'batteryLevel' in device_metrics or 'voltage' in device_metrics or 'channelUtilization' in device_metrics or 'airUtilTx' in device_metrics:
metrics_log.append({'nodeID': fromraw, 'time': rectime, 'battery': device_metrics.get('batteryLevel', 0), 'voltage': round(device_metrics.get('voltage', 0.00),2), 'utilization': round(device_metrics.get('channelUtilization', 0.00),2), 'airutiltx': round(device_metrics.get('airUtilTx', 0.00),2)})
power_metrics = telemetry.get('powerMetrics', {})
if power_metrics:
text_raws += '\n' + (' ' * 11) + 'CH1 Voltage: ' + str(round(power_metrics.get('ch1_voltage', 'N/A'),2)) + 'v'
text_raws += ' CH1 Current: ' + str(round(power_metrics.get('ch1_current', 'N/A'),2)) + 'mA'
text_raws += ' CH2 Voltage: ' + str(round(power_metrics.get('ch2_voltage', 'N/A'),2)) + 'v'
text_raws += ' CH2 Current: ' + str(round(power_metrics.get('ch2_current', 'N/A'),2)) + 'mA'
environment_metrics = telemetry.get('environmentMetrics', {})
if environment_metrics:
text_raws += '\n' + (' ' * 11) + 'Temperature: ' + str(round(environment_metrics.get('temperature', 0.0),1)) + '°C'
text_raws += ' Humidity: ' + str(round(environment_metrics.get('relativeHumidity', 0.0),1)) + '%'
text_raws += ' Pressure: ' + str(round(environment_metrics.get('barometricPressure', 0.00),2)) + 'hPa'
if 'temperature' in environment_metrics or 'relativeHumidity' in environment_metrics or 'barometricPressure' in environment_metrics:
environment_log.append({'nodeID': fromraw, 'time': rectime, 'temperature': round(environment_metrics.get('temperature', 0.0),2), 'humidity': round(environment_metrics.get('relativeHumidity', 0.0),2), 'pressure': round(environment_metrics.get('barometricPressure', 0.00),2)})
localstats_metrics = telemetry.get('localStats', {})
if localstats_metrics:
text_raws += '\n' + (' ' * 11) + 'PacketsTx: ' + str(localstats_metrics.get('numPacketsTx', 0))
text_raws += ' PacketsRx: ' + str(localstats_metrics.get('numPacketsRx', 0))
text_raws += ' PacketsRxBad: ' + str(localstats_metrics.get('numPacketsRxBad', 0))
if device_metrics.get('numTxRelay', 0) > 0:
text_raws += '\n' + (' ' * 11) + 'TxRelay: ' + str(localstats_metrics.get('numTxRelay', 0))
if device_metrics.get('numRxDupe', 0) > 0:
text_raws += ' RxDupe: ' + str(localstats_metrics.get('numRxDupe', 0))
if device_metrics.get('numTxRelayCanceled', 0) > 0:
text_raws += ' TxCanceled: ' + str(localstats_metrics.get('numTxRelayCanceled', 0))
text_raws += ' Nodes: ' + str(localstats_metrics.get('numOnlineNodes', 0)) + '/' + str(localstats_metrics.get('numTotalNodes', 0))
if MyLora == fromraw:
MyLoraText2 = (' PacketsTx').ljust(13) + str(localstats_metrics.get('numPacketsTx', 0)).rjust(7) + '\n' + (' PacketsRx').ljust(13) + str(localstats_metrics.get('numPacketsRx', 0)).rjust(7) + '\n' + (' Rx Bad').ljust(13) + str(localstats_metrics.get('numPacketsRxBad', 0)).rjust(7) + '\n' + (' Nodes').ljust(13) + (str(localstats_metrics.get('numOnlineNodes', 0)) + '/' + str(localstats_metrics.get('numTotalNodes', 0))).rjust(7) + '\n'
if text_raws == 'Node Telemetry':
text_raws += ' No Data'
elif data["portnum"] == "CHAT_APP" or data["portnum"] == "TEXT_MESSAGE_APP":
text = ''
if 'chat' in data:
text = data.get('chat', '')
if 'text' in data:
text = data.get('text', '')
if text != '':
text_msgs = str(text.encode('ascii', 'xmlcharrefreplace'), 'ascii').rstrip()
text_raws = text
text_chns = 'Private'
if "toId" in packet:
if packet["toId"] == '^all':
text_chns = text_chns = str(mylorachan[0])
if "channel" in packet:
text_chns = str(mylorachan[packet["channel"]])
ischat = True
playsound('Data' + os.path.sep + 'NewChat.mp3')
else:
text_raws = 'Node Chat Encrypted'
elif data["portnum"] == "POSITION_APP":
position = data["position"]
nodelat = round(position.get('latitude', -8.0),6)
nodelon = round(position.get('longitude', -8.0),6)
text_msgs = 'Node Position '
text_msgs += 'latitude ' + str(round(nodelat,4)) + ' '
text_msgs += 'longitude ' + str(round(nodelon,4)) + ' '
text_msgs += 'altitude ' + str(position.get('altitude', 0)) + ' meter\n' + (' ' * 11)
if nodelat != -8.0 and nodelon != -8.0:
logLora(fromraw, ['POSITION_APP', nodelat, nodelon, position.get('altitude', 0)])
if MyLora != fromraw and LoraDB[fromraw][3] != -8.0 and LoraDB[fromraw][4] != -8.0:
text_msgs += "Distance: ±" + calc_gc(nodelat, nodelon, LoraDB[MyLora][3], LoraDB[MyLora][4]) + " "
if fromraw in MapMarkers and MapMarkers[fromraw][0] != None:
MapMarkers[fromraw][0].set_position(nodelat, nodelon)
MapMarkers[fromraw][0].set_text(LoraDB[fromraw][1])
last_position = get_last_position(movement_log, fromraw)
if last_position and 'latitude' in position and 'longitude' in position:
if last_position['latitude'] != nodelat or last_position['longitude'] != nodelon:
MapMarkerDelete(fromraw)
movement_log.append({'nodeID': fromraw, 'time': rectime, 'latitude': nodelat, 'longitude': nodelon, 'altitude': position.get('altitude', 0)})
text_msgs += '(Moved!) '
if 'precisionBits' in position and position.get('precisionBits', 0) > 0:
AcMeters = round(23905787.925008 * math.pow(0.5, position.get('precisionBits', 0)), 2)
if AcMeters > 1.0:
text_msgs += '(Accuracy ±' + print_range(AcMeters) + ') '
if fromraw in MapMarkers and AcMeters >= 30.0 and AcMeters <= 5000.0:
# Lets draw only a circle if distance bigger then 30m or smaller then 5km
if len(MapMarkers[fromraw]) == 7:
MapMarkers[fromraw].append(None)
MapMarkers[fromraw][7] = mapview.set_polygon(position=(nodelat, nodelon), range_in_meters=(AcMeters * 2),fill_color="gray25")
if not last_position and 'latitude' in position and 'longitude' in position:
movement_log.append({'nodeID': fromraw, 'time': rectime, 'latitude': nodelat, 'longitude': nodelon, 'altitude': position.get('altitude', 0)})
if "satsInView" in position:
text_msgs += '(' + str(position.get('satsInView', 0)) + ' satelites)'
text_raws = text_msgs
elif data["portnum"] == "NODEINFO_APP":
node_info = packet['decoded'].get('user', {})
if node_info:
lora_sn = str(node_info.get('shortName', str(fromraw)[:-4]).encode('ascii', 'xmlcharrefreplace'), 'ascii')
lora_ln = str(node_info.get('longName', 'N/A').encode('ascii', 'xmlcharrefreplace'), 'ascii')
lora_mc = node_info.get('macaddr', 'N/A')
lora_mo = node_info.get('hwModel', 'N/A')
logLora(fromraw, ['NODEINFO_APP', lora_sn, lora_ln, lora_mc, lora_mo])
if fromraw in MapMarkers:
MapMarkers[fromraw][0].set_text(unescape(lora_sn))
text_raws = "Node Info using hardware " + lora_mo
if 'isLicensed' in packet:
text_raws += " (Licensed)"
if 'role' in packet:
text_raws += " Role: " + node_info.get('role', 'N/A')
text_from = lora_sn + " (" + lora_ln + ")"
else:
text_raws = 'Node Info No Data'
elif data["portnum"] == "NEIGHBORINFO_APP":
text_raws = 'Node Neighborinfo'
listmaps = []
if fromraw not in MapMarkers and fromraw in LoraDB:
if LoraDB[fromraw][3] != -8.0 and LoraDB[fromraw][4] != -8.0:
MapMarkers[fromraw] = [None, True, tnow, None, None, 0, None]
MapMarkers[fromraw][0] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], text=unescape(LoraDB[fromraw][1]), icon_index=3, text_color = '#2bd5ff', font = ('Fixedsys', 8), data=fromraw, command = click_command)
if fromraw in MapMarkers:
if len(MapMarkers[fromraw]) > 3 and MapMarkers[fromraw][3] is not None:
MapMarkers[fromraw][3].delete()
MapMarkers[fromraw][3] = None
if "neighborinfo" in data and "neighbors" in data["neighborinfo"]:
text = data["neighborinfo"]["neighbors"]
if fromraw in MapMarkers and MapMarkers[fromraw][3] is not None:
MapMarkers[fromraw][3].delete()
MapMarkers[fromraw][3] = None
for neighbor in text:
nodeid = hex(neighbor["nodeId"])[2:]
if nodeid in LoraDB and LoraDB[nodeid][1] != '':
LoraDB[nodeid][0] = tnow
# Lets add to map ass well if we are not on map abd our db knows the station
if nodeid not in MapMarkers:
if LoraDB[nodeid][3] != -8.0 and LoraDB[nodeid][4] != -8.0:
MapMarkers[nodeid] = [None, True, tnow, None, None, 0, None]
MapMarkers[nodeid][0] = mapview.set_marker(LoraDB[nodeid][3], LoraDB[nodeid][4], text=unescape(LoraDB[nodeid][1]), icon_index=3, text_color = '#2bd5ff', font = ('Fixedsys', 8), data=fromraw, command = click_command)
else:
MapMarkers[nodeid][2] = tnow
# Lets add to paths ass well if we are on map
if fromraw in MapMarkers:
if LoraDB[nodeid][3] != -8.0 and LoraDB[nodeid][4] != -8.0:
listmaps = []
pos = (LoraDB[fromraw][3], LoraDB[fromraw][4])
listmaps.append(pos)
pos = (LoraDB[nodeid][3], LoraDB[nodeid][4])
listmaps.append(pos)
MapMarkers[fromraw][3] = mapview.set_path(listmaps, color="#006642", width=2)
nodeid = LoraDB[nodeid][1]
else:
nodeid = '!' + nodeid
text_raws += '\n' + (' ' * 11) + nodeid
if "snr" in neighbor:
text_raws += ' (' + str(neighbor["snr"]) + 'dB)'
else:
text_raws += ' No Data'
elif data["portnum"] == "RANGE_TEST_APP":
text_raws = 'Node RangeTest'
payload = data.get('payload', b'')
text_raws += '\n' + (' ' * 11) + 'Payload: ' + str(payload.decode())
elif data["portnum"] == "TRACEROUTE_APP":
TraceTo = idToHex(packet['to'])
TraceFrom = idToHex(packet['from'])
if TraceTo[1:] in LoraDB: TraceTo = LoraDB[TraceTo[1:]][1]
if TraceFrom[1:] in LoraDB: TraceFrom = LoraDB[TraceFrom[1:]][1]
route = packet['decoded']['traceroute'].get('route', [])
snr = packet['decoded']['traceroute'].get('snrTowards', [])
routeBack = packet['decoded']['traceroute'].get('routeBack', [])
snrBack = packet['decoded']['traceroute'].get('snrBack', [])
text_raws = 'Node Traceroute\n' + (' ' * 11) + 'From : ' + TraceTo + ' --> '
index = 0
if routeBack:
for nodeuuid in routeBack:
nodeidt = idToHex(nodeuuid)[1:]
if nodeidt in LoraDB:
text_raws += LoraDB[nodeidt][1]
else:
text_raws += '!' + nodeidt
if snrBack and snrBack[index] != -128 and snrBack[index] != 0:
text_raws += f" ({snrBack[index] / 4:.2f}dB)"
text_raws += ' --> '
index += 1
text_raws += TraceFrom
if snrBack and snrBack[index] != -128 and snrBack[index] != 0:
text_raws += f" ({snrBack[index] / 4:.2f}dB)"
text_raws += '\n' + (' ' * 11) + 'Back : ' + TraceFrom + ' --> '
index = 0
if route:
for nodeuuid in route:
nodeidt = idToHex(nodeuuid)[1:]
if nodeidt in LoraDB:
text_raws += LoraDB[nodeidt][1]
else:
text_raws += '!' + nodeidt
if snr and snr[index] != -128 and snr[index] != 0:
text_raws += f" ({snr[index] / 4:.2f}dB)"
text_raws += ' --> '
index += 1
text_raws += TraceTo
if snr and snr[index] != -128 and snr[index] != 0:
text_raws += f" ({snr[index] / 4:.2f}dB)"
elif data["portnum"] == "ROUTING_APP":
text_raws = 'Node Routing'
if "errorReason" in data["routing"]:
text_raws += ' - Error : ' + data["routing"]["errorReason"]
else:
# Unknown Packet
if 'portnum' in data:
text_raws = 'Node ' + (data["portnum"].split('_APP', 1)[0]).title()
else:
text_raws = 'Node Unknown Packet'
if "rxSnr" in packet and packet['rxSnr'] is not None:
# we want rxRssi / rxSnr
LoraDB[fromraw][11] = str(packet['rxSnr']) + 'dB'
# Lets work the map
if fromraw != MyLora:
if fromraw in MapMarkers:
MapMarkers[fromraw][2] = tnow
if viaMqtt == True and MapMarkers[fromraw][1] == False:
MapMarkers[fromraw][1] = True
if MapMarkers[fromraw][0] != None:
MapMarkers[fromraw][0].change_icon(3)
elif viaMqtt == False and MapMarkers[fromraw][1] == True:
MapMarkers[fromraw][1] = False
if MapMarkers[fromraw][0] != None:
MapMarkers[fromraw][0].change_icon(2)
elif LoraDB[fromraw][3] != -8.0 and LoraDB[fromraw][4] != -8.0 and viaMqtt == True:
MapMarkers[fromraw] = [None, True, tnow, None, None, 0, None]
MapMarkers[fromraw][0] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], text=unescape(LoraDB[fromraw][1]), icon_index=3, text_color = '#2bd5ff', font = ('Fixedsys', 8), data=fromraw, command = click_command)
MapMarkers[fromraw][0].text_color = '#2bd5ff'
elif LoraDB[fromraw][3] != -8.0 and LoraDB[fromraw][4] != -8.0 and viaMqtt == False:
MapMarkers[fromraw] = [None, False, tnow, None, None, 0, None]
MapMarkers[fromraw][0] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], text=unescape(LoraDB[fromraw][1]), icon_index=2, text_color = '#2bd5ff', font = ('Fixedsys', 8), data=fromraw, command = click_command)
MapMarkers[fromraw][0].text_color = '#2bd5ff'
# Lets add a indicator
if fromraw in MapMarkers and MapMarkers[fromraw][6] == None and 'localstats_metrics' not in packet:
MapMarkers[fromraw][6] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], icon_index=5, data=fromraw, command = click_command)
# Cleanup and get ready to print
text_from = unescape(text_from)
text_raws = unescape(text_raws)
if text_raws != '' and MyLora != fromraw:
insert_colored_text(text_box1, '[' + time.strftime("%H:%M:%S", time.localtime()) + '] ' + text_from + ' [!' + fromraw + ']' + LoraDB[fromraw][10] + "\n", "#d1d1d1", tag=fromraw)
if ischat == True:
add_message(text_box3, fromraw, text_raws, tnow, private=text_chns)
if viaMqtt == True:
insert_colored_text(text_box1, (' ' * 11) + text_raws + '\n', "#c9a500")
else:
text_from = ''
if LoraDB[fromraw][12] > 0:
text_from = '\n' + (' ' * 11) + str(LoraDB[fromraw][12]) + ' hops '
if LoraDB[fromraw][11] != '' and MyLora != fromraw:
if text_from == '':
text_from = '\n' + (' ' * 11)
v = float(LoraDB[fromraw][11].replace('dB', ''))
text_from += f"{round(v,1)}dB {value_to_graph(v)}"
insert_colored_text(text_box1, (' ' * 11) + text_raws + text_from + '\n', "#00c983")
elif text_raws != '' and MyLora == fromraw:
insert_colored_text(text_box2, "[" + time.strftime("%H:%M:%S", time.localtime()) + '] ' + text_from + LoraDB[fromraw][10] + "\n", "#d1d1d1")
insert_colored_text(text_box2, (' ' * 11) + text_raws + '\n', "#00c983")
else:
insert_colored_text(text_box1, '[' + time.strftime("%H:%M:%S", time.localtime()) + '] ' + text_from + ' [!' + fromraw + ']' + LoraDB[fromraw][10] + "\n", "#d1d1d1", tag=fromraw)
else:
logging.debug("No fromId in packet")
insert_colored_text(text_box1, '[' + time.strftime("%H:%M:%S", time.localtime()) + '] No fromId in packet\n', "#c24400")
else:
if text_from != '':
if text_from in LoraDB:
LoraDB[text_from][0] = tnow
text_from = LoraDB[text_from][1] + " (" + LoraDB[text_from][2] + ") [!" + fromraw + "]"
else:
text_from = "Unknown Node [!" + fromraw + "]"
insert_colored_text(text_box1, '[' + time.strftime("%H:%M:%S", time.localtime()) + ']', "#d1d1d1")
insert_colored_text(text_box1, ' Encrypted packet from ' + text_from + '\n', "#db6544", tag=fromraw)
if fromraw not in MapMarkers and fromraw in LoraDB:
if LoraDB[fromraw][3] != -8.0 and LoraDB[fromraw][4] != -8.0:
MapMarkers[fromraw] = [None, False, tnow, None, None, 0, None]
MapMarkers[fromraw][0] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], text=unescape(LoraDB[fromraw][1]), icon_index=4, text_color = '#aaaaaa', font = ('Fixedsys', 8), data=fromraw, command = click_command)
MapMarkers[fromraw][0].text_color = '#aaaaaa'
MapMarkers[fromraw][6] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], icon_index=5, data=fromraw, command = click_command)
elif fromraw in MapMarkers and MapMarkers[fromraw][0] == None:
MapMarkers[fromraw][6] = mapview.set_marker(LoraDB[fromraw][3], LoraDB[fromraw][4], icon_index=5, data=fromraw, command = click_command)
def updatesnodes():
global LoraDB, MyLora, MapMarkers
info = ''
itmp = 0
tnow = int(time.time())
# a_while_back = tnow - int(timedelta(minutes=5).total_seconds())
for nodes, info in meshtastic_client.nodes.items():
if "user" in info:
tmp = info['user']
if "id" in tmp and tmp['id'] != '':
# Only push to DB if we actually get a node ID
nodeID = str(tmp['id'])[1:]
if nodeID != '':
nodeLast = tnow
itmp = itmp + 1
if "lastHeard" in info and info["lastHeard"] is not None:
nodeLast = info['lastHeard']
if nodeID not in LoraDB:
LoraDB[nodeID] = [nodeLast, nodeID[-4:], '', -8.0, -8.0, 0, '', '', nodeLast, '0% 0.0v', '', '',-1, 0]
insert_colored_text(text_box1, "[" + time.strftime("%H:%M:%S", time.localtime()) + "]", "#d1d1d1")
insert_colored_text(text_box1, " New Node Logged [!" + nodeID + "]\n", "#e8643f", tag=nodeID)
if "shortName" in tmp and "longName" in tmp:
lora_sn = str(tmp['shortName'].encode('ascii', 'xmlcharrefreplace'), 'ascii').replace("\n", "")
lora_ln = str(tmp['longName'].encode('ascii', 'xmlcharrefreplace'), 'ascii').replace("\n", "")
if lora_sn in lora_ln and "Meshtastic" in lora_ln:
if LoraDB[nodeID][1] == '': LoraDB[nodeID][1] = lora_ln
if LoraDB[nodeID][2] == '': LoraDB[nodeID][2] = lora_ln
else:
LoraDB[nodeID][1] = lora_sn
LoraDB[nodeID][2] = lora_ln
if "macaddr" in tmp: LoraDB[nodeID][6] = str(tmp['macaddr'])
if "hwModel" in tmp: LoraDB[nodeID][7] = str(tmp['hwModel'])
LoraDB[nodeID][12] = -1
if "hopsAway" in info: LoraDB[nodeID][12] = info['hopsAway']
if "position" in info and LoraDB[nodeID][3] == -8.0 and LoraDB[nodeID][4] == -8.0:
tmp2 = info['position']
if "latitude" in tmp2 and "longitude" in tmp2 and tmp2['latitude'] != '' and tmp2['longitude'] != '':
LoraDB[nodeID][3] = round(tmp2.get('latitude', -8.0),6)
LoraDB[nodeID][4] = round(tmp2.get('longitude', -8.0),6)
if "altitude" in tmp:
LoraDB[nodeID][5] = tmp2.get('altitude', 0)
if nodeID == MyLora:
LoraDB[MyLora][0] = tnow
if LoraDB[MyLora][3] != -8.0 and LoraDB[MyLora][4] != -8.0:
if MyLora not in MapMarkers:
mapview.set_zoom(11)
MapMarkers[MyLora] = [None, False, tnow, None, None, 0, None]
MapMarkers[MyLora][0] = mapview.set_marker(LoraDB[MyLora][3], LoraDB[MyLora][4], text=unescape(LoraDB[MyLora][1]), icon_index=1, text_color = '#e67a7f', font = ('Fixedsys', 8), data=MyLora, command = click_command)
MapMarkers[MyLora][0].text_color = '#e67a7f'
mapview.set_position(LoraDB[MyLora][3], LoraDB[MyLora][4])
else:
insert_colored_text(text_box2, "[" + time.strftime("%H:%M:%S", time.localtime()) + "]", "#d1d1d1")
insert_colored_text(text_box2, " My Node has no position !!\n", "#e8643f")
if "viaMqtt" in info: LoraDB[nodeID][10] = ' via mqtt'
if "snr" in info and info['snr'] is not None: LoraDB[nodeID][11] = str(info['snr']) + 'dB'
#-------------------------------------------------------------- Side Functions ---------------------------------------------------------------------------
def ez_date(d):
ts = d
if ts > 31536000:
temp = int(round(ts / 31536000, 0))
val = f"{temp} year{'s' if temp > 1 else ''}"
elif ts > 2419200:
temp = int(round(ts / 2419200, 0))
val = f"{temp} month{'s' if temp > 1 else ''}"
elif ts > 604800:
temp = int(round(ts / 604800, 0))
val = f"{temp} week{'s' if temp > 1 else ''}"
elif ts > 86400:
temp = int(round(ts / 86400, 0))
val = f"{temp} day{'s' if temp > 1 else ''}"
elif ts > 3600:
temp = int(round(ts / 3600, 0))
val = f"{temp} hour{'s' if temp > 1 else ''}"
elif ts > 60:
temp = int(round(ts / 60, 0))
val = f"{temp} minute{'s' if temp > 1 else ''}"
else:
temp = int(ts)
val = "Just now"
return val
def uptimmehuman(node_id):
tnow = int(time.time())
days, remainder = divmod(LoraDB[node_id][13] + (tnow - LoraDB[node_id][0]), 86400)
hours, remainder = divmod(remainder, 3600)
minutes, seconds = divmod(remainder, 60)
text = 'Uptime : '
if days > 0: text += str(days) + ' days, '
text += str(hours) + ' hours and ' + str(minutes) + ' minutes'
if tnow - LoraDB[node_id][0] >= map_delete: text += ' ? Seems offline'
return text
def LatLon2qth(latitude, longitude):
A = ord('A')
a = divmod(longitude + 180, 20)
b = divmod(latitude + 90, 10)
locator = chr(A + int(a[0])) + chr(A + int(b[0]))
lon = a[1] / 2.0
lat = b[1]
i = 1
while i < 5:
i += 1
a = divmod(lon, 1)
b = divmod(lat, 1)
if not (i % 2):
locator += str(int(a[0])) + str(int(b[0]))
lon = 24 * a[1]
lat = 24 * b[1]
else:
locator += chr(A + int(a[0])) + chr(A + int(b[0]))
lon = 10 * a[1]
lat = 10 * b[1]
return locator
def calc_gc(end_lat, end_long, start_lat, start_long):
start_lat = math.radians(start_lat)
start_long = math.radians(start_long)
end_lat = math.radians(end_lat)
end_long = math.radians(end_long)
d_lat = math.fabs(start_lat - end_lat)
d_long = math.fabs(start_long - end_long)
EARTH_R = 6372.8
y = ((math.sin(start_lat)*math.sin(end_lat)) + (math.cos(start_lat)*math.cos(end_lat)*math.cos(d_long)))
x = math.sqrt((math.cos(end_lat)*math.sin(d_long))**2 + ( (math.cos(start_lat)*math.sin(end_lat)) - (math.sin(start_lat)*math.cos(end_lat)*math.cos(d_long)))**2)
c = math.atan(x/y)
return f"{round(EARTH_R*c,1)}km"
#-------------------------------------------------------------- Plot Functions ---------------------------------------------------------------------------
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.ticker import ScalarFormatter
from pandas import DataFrame
from scipy.signal import savgol_filter
plt.switch_backend('TkAgg') # No clue why we even need this
plt.rcParams["font.family"] = 'sans-serif'
plt.rcParams["font.size"] = 7
def plot_metrics_log(metrics_log, node_id, frame, width=512, height=212):
global MyLora
metrics = get_data_for_node(metrics_log, node_id)
df = DataFrame({
'time': [datetime.fromtimestamp(entry['time']) for entry in metrics],
'battery': [entry['battery'] for entry in metrics],
'voltage': [entry['voltage'] for entry in metrics],
'utilization': [entry['utilization'] for entry in metrics],
'airutiltx': [entry['airutiltx'] for entry in metrics]
})
resample_interval = len(df) // 60 or 5
df_resampled = df.set_index('time').resample(f'{resample_interval}min').mean().dropna().reset_index()
times_resampled = df_resampled['time'].tolist()
battery_levels_resampled = df_resampled['battery'].tolist()
voltages_resampled = df_resampled['voltage'].tolist()
utilizations_resampled = df_resampled['utilization'].tolist()
airutiltxs_resampled = df_resampled['airutiltx'].tolist()
if len(battery_levels_resampled) < 5 or len(voltages_resampled) < 5 or len(utilizations_resampled) < 5 or len(airutiltxs_resampled) < 5:
return
battery_levels_smooth = savgol_filter(battery_levels_resampled, window_length=5, polyorder=2)
voltages_smooth = savgol_filter(voltages_resampled, window_length=5, polyorder=2)
utilizations_smooth = savgol_filter(utilizations_resampled, window_length=5, polyorder=2)
airutiltxs_smooth = savgol_filter(airutiltxs_resampled, window_length=5, polyorder=2)
total_hours = 0
if len(times_resampled) > 1:
total_hours = (times_resampled[-1] - times_resampled[0]).total_seconds() / 3600
fig, axs = plt.subplots(2, 2, figsize=(width/100, height/100))
fig.patch.set_facecolor('#242424')
# Plot battery levels
axs[0, 0].plot(times_resampled, battery_levels_smooth, label='Battery Level', color='#2bd5ff')
axs[0, 0].set_title('Battery Level %')
axs[0, 0].set_xlabel(None)
axs[0, 0].set_ylabel(None)
axs[0, 0].grid(True, color='#444444')
# Plot voltages
axs[0, 1].plot(times_resampled, voltages_smooth, label='Voltage', color='#c9a500')
axs[0, 1].set_title('Voltage')
axs[0, 1].set_xlabel(None)
axs[0, 1].set_ylabel(None)
axs[0, 1].grid(True, color='#444444')
# Plot utilizations
axs[1, 0].plot(times_resampled, utilizations_smooth, label='Utilization', color='#00c983')
axs[1, 0].set_title('Utilization %')
axs[1, 0].set_xlabel(None)
axs[1, 0].set_ylabel(None)
axs[1, 0].grid(True, color='#444444')
# Plot Air Utilization TX
axs[1, 1].plot(times_resampled, airutiltxs_smooth, label='Air Utilization TX', color='#ee0000')
axs[1, 1].set_title('Air Utilization TX %')
axs[1, 1].set_xlabel(None)
axs[1, 1].set_ylabel(None)
axs[1, 1].grid(True, color='#444444')
for ax in axs.flat:
ax.set_facecolor('#242424')
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H'))
if total_hours > 11:
ax.xaxis.set_major_locator(mdates.HourLocator(interval=3))
else:
ax.xaxis.set_major_locator(mdates.HourLocator())