-
-
Notifications
You must be signed in to change notification settings - Fork 14
/
main.py
executable file
·1574 lines (1207 loc) · 114 KB
/
main.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
# Predator
# Copyright (C) 2024 V0LT - Conner Vieira
# This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by# the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License along with this program (LICENSE)
# If not, see https://www.gnu.org/licenses/ to read the license agreement.
print("Loading Predator...")
import os # Required to interact with certain operating system functions
import json # Required to process JSON data
import config
load_config = config.load_config
validate_config = config.validate_config
config = load_config()
predator_root_directory = str(os.path.dirname(os.path.realpath(__file__))) # This variable determines the folder path of the root Predator directory. This should usually automatically recognize itself, but it if it doesn't, you can change it manually.
import time # Required to add delays and handle dates/times.
import sys # Required to read command line arguments.
import signal # Required to kill Predator.
import utils # Import the utils.py scripts.
style = utils.style # Load the style from the utils script.
debug_message = utils.debug_message # Load the debug message function from the utils script.
clear = utils.clear # Load the screen clearing function from the utils script.
prompt = utils.prompt # Load the user input prompt function from the utils script.
is_json = utils.is_json # Load the function used to determine if a given string is valid JSON.
play_sound = utils.play_sound # Load the function used to play sounds from the utils script.
display_message = utils.display_message # Load the message display function from the utils script.
process_gpx = utils.process_gpx # Load the GPX processing function from the utils script.
save_to_file = utils.save_to_file # Load the file saving function from the utils script.
add_to_file = utils.add_to_file # Load the file appending function from the utils script.
display_shape = utils.display_shape # Load the shape displaying function from the utils script.
countdown = utils.countdown # Load the timer countdown function from the utils script.
get_gps_location = utils.get_gps_location # Load the function to get the current GPS location.
convert_speed = utils.convert_speed # Load the function used to convert speeds from meters per second to other units.
display_number = utils.display_number # Load the function used to display numbers as large ASCII font.
closest_key = utils.closest_key # Load the function used to find the closest entry in a dictionary to a given number.
heartbeat = utils.heartbeat # Load the function to issue heartbeats to the interface directory.
update_state = utils.update_state # Load the function to issue state updates to the interface directory.
log_plates = utils.log_plates # Load the function to issue ALPR results to the interface directory.
log_alerts = utils.log_alerts # Load the function to issue active alerts to the interface directory.
import re # Required to use Regex.
import datetime # Required for converting between timestamps and human readable date/time information.
import fnmatch # Required to use wildcards to check strings.
if (config["general"]["modes"]["enabled"]["realtime"] == True):
import alpr
display_alerts = alpr.display_alerts # Load the function used to display license plate alerts given the dictionary of alerts.
load_alert_database = alpr.load_alert_database # Load the function used to load license plate alert databases.
if (config["general"]["modes"]["enabled"]["dashcam"] == True): # Check to see if OpenCV is needed.
import dashcam
if (config["developer"]["offline"] == False): # Only import networking libraries if offline mode is turned off.
if (config["general"]["status_lighting"]["enabled"] == True or config["realtime"]["push_notifications"]["enabled"] == True or len(config["general"]["alerts"]["databases"]) > 0): # Only import networking libraries if they are necessary.
debug_message("Loading networking libraries")
import requests # Required to make network requests.
import validators # Required to validate URLs.
if (config["management"]["disk_statistics"] == True): # Only import the disk statistic library if it is enabled in the configuration.
debug_message("Loading system utility library")
import psutil # Required to get disk usage information
if (os.path.exists(predator_root_directory + "/install.json") == False): # Check to see if the install information file hasn't yet been created. This will be the case on the first start-up.
import uuid
install_data = {"first_start_time": int(time.time()), "id": str(uuid.uuid4())}
clear()
print(style.bold + style.red + "Predator - First Start" + style.end)
print(style.italic + "This wizard is only displayed on the first start of Predator.")
print("To reset so this message is displayed again, remove the `install.json` file inside the main install directory." + style.end)
input(style.faint + "Press enter to continue..." + style.end)
clear()
print(style.bold + style.red + "Commercial Support" + style.end)
print(style.bold + "V0LT offers the following commercial support services for Predator:" + style.end)
print(" * Custom software modifications")
print(" * Pre-assembled hardware products and kits")
print(" * One-on-one technical support")
print(" * Server hosting")
print(" - Remotely-managed hot-lists and ignore-lists")
print(" - Compiled data reports")
print(" - Remote ALPR processing")
print()
print("To learn more, don't hesitate to get in contact: " + style.underline + "https://v0lttech.com/contact.php\n" + style.end)
input(style.faint + "Press enter to continue..." + style.end)
clear()
print(style.bold + style.red + "Warranty" + style.end)
print("While Predator is designed to be as reliable and consistent as possible, it comes with absolutely no warranty, it should not be used in a context where failure could cause harm to people or property.")
print("For more information, see the `SECURITY.md` document.")
input(style.faint + "Press enter to continue..." + style.end)
clear()
print(style.bold + style.red + "Privacy" + style.end)
print("Predator does not share telemetry or usage data with V0LT, or any other entity. However, by default, Predator will attach a random identifier to requests made to remote license plate list sources (as configured under `general>alerts>databases`). This identifier allows administrators of servers hosting license plate lists to roughly count how many clients are using their lists. If you're concerned about the administrator of one of your remote license plate lists using this unique identifier to derive information about how often you use Predator (based on when you fetch their lists), you can disable this functionality using the `developer>identify_to_remote_sources` configuration value.")
print("Additionally, by default, Predator will fetch a hardcoded ignore list from the V0LT website. Once again, this functionality does not send any identifiable information or telemetry data. To disable this functionality, either enable the `developer>offline` configuration value to disable all network requests, or remove the hardcoded ignore list from `ignore.py`.")
print("For more information, see the `docs/CONFIGURE.md` document.")
input(style.faint + "Press enter to continue..." + style.end)
clear()
print(style.bold + style.red + "Funding" + style.end)
print("Predator is completely free to use, and doesn't contain monetization like advertising or sponsorships. If you find the project to be useful, please consider supporting it financially.")
print("For more information, see `https://v0lttech.com/donate.php`.")
input(style.faint + "Press enter to continue..." + style.end)
with open(predator_root_directory + "/install.json", 'w') as file:
json.dump(install_data, file)
del install_data
print("")
clear()
print("The initial start-up process has completed. Predator will now return to the normal start-up sequence.")
input(style.faint + "Press enter to continue..." + style.end)
debug_message("Loading ignore lists")
import ignore # Import the library to handle license plates in the ignore list.
ignore_list = ignore.fetch_ignore_list() # Fetch the ignore lists.
debug_message("Validating configuration values")
invalid_configuration_values = validate_config(config) # Validation the configuration, and display any potential problems.
for entry in invalid_configuration_values: # Iterate through each invalid configuration value in the list.
display_message("Invalid configuration value: " + entry, 3) # Print each invalid configuration value as an error.
del invalid_configuration_values # Delete the variable holding the list of invalid configuration_values.
debug_message("Validated configuration values")
if (config["developer"]["offline"] == True): # If offline mode is enabled, then disable all network based features.
config["realtime"]["push_notifications"]["enabled"] = False
config["realtime"]["push_notifications"]["server"] = "" # This is redundant, since 'realtime>push_notifications>enabled' is disabled, but it serves as a backup.
config["general"]["status_lighting"]["enabled"] = False
config["developer"]["remote_sources"] = []
heartbeat() # Issue an initial heartbeat at start-up.
if (config["general"]["object_recognition"]["enabled"] == True): # Check to see whether object recognition (Tensorflow/OpenCV) is enabled.
debug_message("Loading object recognition")
try: # "Try" to import Tensorflow and OpenCV; Don't quit the entire program if an error is encountered.
import silence_tensorflow.auto # Silences tensorflow warnings
import cv2 # Required for object recognition (not license plate recognition)
import cvlib as cv # Required for object recognition (not license plate recognition)
from cvlib.object_detection import draw_bbox # Required for object recognition (not license plate recognition)
except Exception:
display_message("The object recognition libraries could not be imported.", 3)
import lighting # Import the lighting.py script.
update_status_lighting = lighting.update_status_lighting # Load the status lighting update function from the lighting script.
debug_message("Initial loading complete")
# Display the start-up intro header.
clear()
if (config["general"]["display"]["ascii_art_header"] == True): # Check to see whether the user has configured there to be a large ASCII art header, or a standard text header.
print(style.red + style.bold)
print(" /$$$$$$$ /$$$$$$$ /$$$$$$$$ /$$$$$$$ /$$$$$$ /$$$$$$$$ /$$$$$$ /$$$$$$$ ")
print("| $$__ $$| $$__ $$| $$_____/| $$__ $$ /$$__ $$|__ $$__//$$__ $$| $$__ $$")
print("| $$ \ $$| $$ \ $$| $$ | $$ \ $$| $$ \ $$ | $$ | $$ \ $$| $$ \ $$")
print("| $$$$$$$/| $$$$$$$/| $$$$$ | $$ | $$| $$$$$$$$ | $$ | $$ | $$| $$$$$$$/")
print("| $$____/ | $$__ $$| $$__/ | $$ | $$| $$__ $$ | $$ | $$ | $$| $$__ $$")
print("| $$ | $$ \ $$| $$ | $$ | $$| $$ | $$ | $$ | $$ | $$| $$ \ $$")
print("| $$ | $$ | $$| $$$$$$$$| $$$$$$$/| $$ | $$ | $$ | $$$$$$/| $$ | $$")
print("|__/ |__/ |__/|________/|_______/ |__/ |__/ |__/ \______/ |__/ |__/" + style.end + style.bold)
print("")
print(" COMPUTER VISION")
if (config["general"]["display"]["startup_message"] != ""): # Only display the line for the custom message if the user has defined one.
print("")
print(config["general"]["display"]["startup_message"]) # Show the user's custom defined start-up message.
print(style.end)
else: # If the user his disabled the large ASCII art header, then show a simple title header with minimal styling.
print(style.red + style.bold + "PREDATOR" + style.end)
print(style.bold + "Computer Vision" + style.end + "\n")
if (config["general"]["display"]["startup_message"]!= ""): # Only display the line for the custom message if the user has defined one.
print(config["general"]["display"]["startup_message"]) # Show the user's custom defined start-up message.
play_sound("startup")
if (config["realtime"]["push_notifications"]["enabled"] == True): # Check to see if the user has push notifications enabled.
debug_message("Issuing start-up push notification")
os.system("curl -X POST '" + config["realtime"]["push_notifications"]["server"] + "/message?token=" + config["realtime"]["push_notifications"]["token"] + "' -F 'title=Predator' -F 'message=Predator has been started.' > /dev/null 2>&1 &") # Send a push notification via Gotify indicating that Predator has started.
# Run some basic error checks to see if any of the data supplied in the configuration seems wrong.
debug_message("Validating configuration")
config["general"]["alpr"]["engine"] = config["general"]["alpr"]["engine"].lower().strip() # Convert the ALPR engine configuration value to all lowercase, and trim leading and trailing white-spaces.
if (config["general"]["alpr"]["engine"] != "phantom" and config["general"]["alpr"]["engine"] != "openalpr"): # Check to see if the configured ALPR engine is invalid.
display_message("The configured ALPR engine is invalid. Please select either 'phantom' or 'openalpr' in the configuration.", 3)
if (os.path.isdir(config["general"]["working_directory"]) == False): # Check to see if the configured working directory is missing.
display_message("The 'general>working_directory' configuration value does not point to an existing directory.", 3)
elif ("'" in config["general"]["working_directory"]):
display_message("The 'general>working_directory' configuration value contains an apostrophe. This will likely cause unexpected behavior.", 3)
elif ("\"" in config["general"]["working_directory"]):
display_message("The 'general>working_directory' configuration value contains a quotation mark. This will likely cause unexpected behavior.", 3)
if (os.path.isdir(config["general"]["interface_directory"]) == False and config["general"]["interface_directory"] != ""): # Check to see if the configured interface directory is missing.
display_message("The 'general>interface_directory' configuration value does not point to an existing directory.", 3)
elif ("'" in config["general"]["interface_directory"]):
display_message("The 'general>interface_directory' configuration value contains an apostrophe. This will likely cause unexpected behavior.", 3)
elif ("\"" in config["general"]["interface_directory"]):
display_message("The 'general>interface_directory' configuration value contains a quotation mark. This will likely cause unexpected behavior.", 3)
if (config["prerecorded"]["image"]["processing"]["cropping"]["left_margin"] < 0 or config["prerecorded"]["image"]["processing"]["cropping"]["right_margin"] < 0 or config["prerecorded"]["image"]["processing"]["cropping"]["bottom_margin"] < 0 or config["prerecorded"]["image"]["processing"]["cropping"]["top_margin"] < 0): # Check to make sure that all of the pre-recorded mode cropping margins are positive numbers.
display_message("One or more of the cropping margins for pre-recorded mode are below 0. This should never happen, and it's likely there's a configuration issue somewhere. Cropping margins have all been set to 0.", 3)
config["prerecorded"]["image"]["processing"]["cropping"]["left_margin"] = 0
config["prerecorded"]["image"]["processing"]["cropping"]["right_margin"] = 0
config["prerecorded"]["image"]["processing"]["cropping"]["bottom_margin"] = 0
config["prerecorded"]["image"]["processing"]["cropping"]["top_margin"] = 0
for device in config["realtime"]["image"]["camera"]["devices"]: # Iterate through each video device specified in the configuration.
if (os.path.exists(config["realtime"]["image"]["camera"]["devices"][device]) == False): # Check to make sure that a camera device points to a valid file.
display_message("The 'realtime>image>camera>devices>" + device + "' configuration value does not point to a valid file.", 3)
if (config["realtime"]["push_notifications"]["enabled"] == True): # Check to see if the user has Gotify notifications turned on in the configuration.
if (config["realtime"]["push_notifications"]["server"] == "" or config["realtime"]["push_notifications"]["server"] == None): # Check to see if the gotify server configuration value has been left blank
display_message("The 'realtime>push_notifications>enabled' setting is turned on, but the 'realtime>push_notifications>server' hasn't been set. Push notifications have been disabled.", 3)
config["realtime"]["push_notifications"]["enabled"] = False
if (config["realtime"]["push_notifications"]["token"] == "" or config["realtime"]["push_notifications"]["token"] == None): # Check to see if the Gotify application token has been left blank.
display_message("The 'realtime>push_notifications>token' setting is turned on, but the 'realtime>push_notifications>token' hasn't been set. Push notifications have been disabled.", 3)
config["realtime"]["push_notifications"]["enabled"] = False
# Figure out which mode to boot into.
print("Please select an operating mode.")
if (config["general"]["modes"]["enabled"]["management"] == True): # Only show the management mode option if it's enabled in the configuration.
print("0. Management")
if (config["general"]["modes"]["enabled"]["prerecorded"] == True): # Only show the pre-recorded mode option if it's enabled in the configuration.
print("1. Pre-recorded")
if (config["general"]["modes"]["enabled"]["realtime"] == True): # Only show the real-time mode option if it's enabled in the configuration.
print("2. Real-time")
if (config["general"]["modes"]["enabled"]["dashcam"] == True): # Only show the dash-cam mode option if it's enabled in the configuration.
print("3. Dash-cam")
# Check to see if the auto_start_mode configuration value is an expected value. If it isn't execution can continue, but the user will need to manually select what mode Predator should start in.
config["general"]["modes"]["auto_start"] = str(config["general"]["modes"]["auto_start"]) # Make sure the "general>modes>auto_start" configuration value is a string.
if (config["general"]["modes"]["auto_start"] != "" and config["general"]["modes"]["auto_start"] != "0" and config["general"]["modes"]["auto_start"] != "1" and config["general"]["modes"]["auto_start"] != "2" and config["general"]["modes"]["auto_start"]!= "3"):
display_message("The 'auto_start_mode' configuration value isn't properly set. This value should be blank, '0', '1', '2', or '3'. It's possible there's been a typo.", 3)
if (len(sys.argv) > 1): # Check to see if there is at least 1 command line argument.
if (sys.argv[1] == "0" or sys.argv[1] == "1" or sys.argv[1] == "2" or sys.argv[1] == "3"): # Check to see if a mode override was specified in the Predator command arguments.
config["general"]["modes"]["auto_start"] = sys.argv[1] # Set the automatic start mode to the mode specified by the command line argument.
if (len(sys.argv) > 2): # Check to see if there are at least 2 command line arguments.
if (sys.argv[2] != "--headless"):
config["general"]["working_directory"] = str(sys.argv[2]) # Set the working directory to the path specified by the command line argument.
if (config["general"]["modes"]["auto_start"] == "0" and config["general"]["modes"]["enabled"]["management"] == True): # Based on the configuration, Predator will automatically boot into management mode.
print(style.bold + "Automatically starting into management mode." + style.end)
mode_selection = "0"
elif (config["general"]["modes"]["auto_start"] == "1" and config["general"]["modes"]["enabled"]["prerecorded"] == True): # Based on the configuration, Predator will automatically boot into pre-recorded mode.
print(style.bold + "Automatically starting into pre-recorded mode." + style.end)
mode_selection = "1"
elif (config["general"]["modes"]["auto_start"] == "2" and config["general"]["modes"]["enabled"]["realtime"] == True): # Based on the configuration, Predator will automatically boot into real-time mode.
print(style.bold + "Automatically starting into real-time mode." + style.end)
mode_selection = "2"
elif (config["general"]["modes"]["auto_start"] == "3" and config["general"]["modes"]["enabled"]["dashcam"] == True): # Based on the configuration, Predator will automatically boot into dash-cam mode.
print(style.bold + "Automatically starting into dash-cam mode." + style.end)
mode_selection = "3"
else: # No 'auto start mode' has been configured, so ask the user to select manually.
mode_selection = prompt("Selection: ")
# Initial setup has been completed, and Predator will now load into the specified mode.
# Management mode
if (mode_selection == "0" and config["general"]["modes"]["enabled"]["management"] == True): # The user has selected to boot into management mode.
debug_message("Started management mode")
working_directory_input = prompt("Working directory (Default " + config["general"]["working_directory"] + "): ", optional=True, input_type=str)
if (working_directory_input == ""): # If the user leaves the
working_directory_input = config["general"]["working_directory"]
while (os.path.exists(working_directory_input) == False): # Run forever until the user enters a working directory that exists.
display_message("The specified working directory doesn't seem to exist.", 2)
working_directory_input = prompt("Working directory (Default " + config["general"]["working_directory"] + "): ", optional=True, input_type=str)
config["general"]["working_directory"] = working_directory_input
while True:
clear()
print("Please select an option")
print("0. Quit")
print("1. File Management")
print("2. Information")
print("3. Configuration")
selection = prompt("Selection: ", optional=False, input_type=str)
if (selection == "0"): # The user has selected to quit Predator.
break # Break the 'while true' loop to terminate Predator.
elif (selection == "1"): # The user has selected the "File Management" option.
print(" Please select an option")
print(" 0. Back")
print(" 1. View")
print(" 2. Copy")
print(" 3. Delete")
selection = prompt(" Selection: ", optional=False, input_type=str)
if (selection == "0"): # The user has selected to return back to the previous menu.
continue # Do nothing, and just finish this loop.
elif (selection == "1"): # The user has selected the "view files" option.
os.system("find " + config["general"]["working_directory"]) # Run the 'find' command in the working directory.
utils.wait_for_input()
elif (selection == "2"): # The user has selected the "copy files" option.
# Reset all of the file selections to un-selected.
copy_management_configuration = False
copy_prerecorded_processed_frames = False
copy_prerecorded_gpx_files = False
copy_prerecorded_license_plate_analysis_data = False
copy_prerecorded_object_recognition_data = False
copy_prerecorded_license_plate_location_data = False
copy_realtime_license_plate_recognition_data = False
copy_realtime_object_recognition_data = False
copy_dashcam_video = False
while True: # Run the "copy files" selection menu on a loop forever until the user is finished selecting files.
clear() # Clear the console output before each loop.
print("Please select which files to copy")
print("0. Continue")
print("")
print("===== Management Mode =====")
if (copy_management_configuration == True):
print("M1. [X] Configuration files")
else:
print("M1. [ ] Configuration files")
print("")
print("===== Pre-recorded Mode =====")
if (copy_prerecorded_processed_frames == True):
print("P1. [X] Processed video frames")
else:
print("P1. [ ] Processed video frames")
if (copy_prerecorded_gpx_files == True):
print("P2. [X] GPX files")
else:
print("P2. [ ] GPX files")
if (copy_prerecorded_license_plate_analysis_data == True):
print("P3. [X] License plate analysis data files")
else:
print("P3. [ ] License plate analysis data files")
if (copy_prerecorded_object_recognition_data == True):
print("P4. [X] Object recognition data files")
else:
print("P4. [ ] Object recognition data files")
if (copy_prerecorded_license_plate_location_data == True):
print("P5. [X] License plate location data files")
else:
print("P5. [ ] License plate location data files")
print("")
print("===== Real-time Mode =====")
if (copy_realtime_license_plate_recognition_data == True):
print("R1. [X] License plate recognition data files")
else:
print("R1. [ ] License plate recognition data files")
if (copy_realtime_object_recognition_data == True):
print("R2. [X] Object recognition data files")
else:
print("R2. [ ] Object recognition data files")
print("")
print("===== Dash-cam Mode =====")
if (copy_dashcam_video == True):
print("D1. [X] Dash-cam videos")
else:
print("D1. [ ] Dash-cam videos")
print("")
selection = prompt("Selection: ", optional=False, input_type=str) # Prompt the user for a selection.
if (selection == "0"):
break
# Toggle the file indicated by the user.
elif (selection.lower() == "m1"):
copy_management_configuration = not copy_management_configuration
elif (selection.lower() == "p1"):
copy_prerecorded_processed_frames = not copy_prerecorded_processed_frames
elif (selection.lower() == "p2"):
copy_prerecorded_gpx_files = not copy_prerecorded_gpx_files
elif (selection.lower() == "p3"):
copy_prerecorded_license_plate_analysis_data = not copy_prerecorded_license_plate_analysis_data
elif (selection.lower() == "p4"):
copy_prerecorded_object_recognition_data = not copy_prerecorded_object_recognition_data
elif (selection.lower() == "p5"):
copy_prerecorded_license_plate_location_data = not copy_prerecorded_license_plate_location_data
elif (selection.lower() == "r1"):
copy_realtime_license_plate_recognition_data = not copy_realtime_license_plate_recognition_data
elif (selection.lower() == "r2"):
copy_realtime_object_recognition_data = not copy_realtime_object_recognition_data
elif (selection.lower() == "d1"):
copy_dashcam_video = not copy_dashcam_video
# Prompt the user for the copying destination.
copy_destination = "" # Set the copy_destination as a blank placeholder.
while os.path.exists(copy_destination) == False: # Repeatedly ask the user for a valid copy destination until they enter one that is valid.
copy_destination = prompt("Destination path: ", optional=False, input_type=str) # Prompt the user for a destination path.
# Copy the files as per the user's inputs.
print("Copying files...")
if (copy_management_configuration):
os.system("cp " + predator_root_directory + "/config.json " + copy_destination)
if (copy_prerecorded_processed_frames):
os.system("cp -r " + config["general"]["working_directory"] + "/frames " + copy_destination)
if (copy_prerecorded_gpx_files):
os.system("cp " + config["general"]["working_directory"] + "/*.gpx " + copy_destination)
if (copy_prerecorded_license_plate_analysis_data):
os.system("cp " + config["general"]["working_directory"] + "/pre_recorded_license_plate_export.* " + copy_destination)
if (copy_prerecorded_object_recognition_data):
os.system("cp " + config["general"]["working_directory"] + "/pre_recorded_object_detection_export.* " + copy_destination)
if (copy_prerecorded_license_plate_location_data):
os.system("cp " + config["general"]["working_directory"] + "/pre_recorded_location_data_export.* " + copy_destination)
if (copy_realtime_license_plate_recognition_data):
os.system("cp " + config["general"]["working_directory"] + "/real_time_plates* " + copy_destination)
if (copy_realtime_object_recognition_data):
os.system("cp '" + config["general"]["working_directory"] + "/" + config["realtime"]["saving"]["object_recognition"]["file"] + "' " + copy_destination)
if (copy_dashcam_video):
os.system("cp " + config["general"]["working_directory"] + "/predator_dashcam* " + copy_destination)
clear()
print("Files have finished copying.")
elif (selection == "3"): # The user has selected the "delete files" option.
# Reset all of the file selections to un-selected.
delete_management_custom = False
delete_prerecorded_processed_frames = False
delete_prerecorded_gpx_files = False
delete_prerecorded_license_plate_analysis_data = False
delete_prerecorded_object_recognition_data = False
delete_prerecorded_license_plate_location_data = False
delete_realtime_license_plate_recognition_data = False
delete_realtime_object_recognition_data = False
delete_dashcam_video = False
while True: # Run the "delete files" selection menu on a loop forever until the user is finished selecting files.
clear() # Clear the console output before each loop.
print("Please select which files to delete")
print("0. Continue")
print("")
print("===== Management Mode =====")
if (delete_management_custom == True):
print("M1. [X] Custom file-name (Specified in next step)")
else:
print("M1. [ ] Custom file-name (Specified in next step)")
print("")
print("===== Pre-recorded Mode =====")
if (delete_prerecorded_processed_frames == True):
print("P1. [X] Processed video frames")
else:
print("P1. [ ] Processed video frames")
if (delete_prerecorded_gpx_files == True):
print("P2. [X] GPX files")
else:
print("P2. [ ] GPX files")
if (delete_prerecorded_license_plate_analysis_data == True):
print("P3. [X] License plate analysis data files")
else:
print("P3. [ ] License plate analysis data files")
if (delete_prerecorded_object_recognition_data == True):
print("P4. [X] Object recognition data files")
else:
print("P4. [ ] Object recognition data files")
if (delete_prerecorded_license_plate_location_data == True):
print("P5. [X] License plate location data files")
else:
print("P5. [ ] License plate location data files")
print("")
print("===== Real-time Mode =====")
if (delete_realtime_license_plate_recognition_data == True):
print("R1. [X] License plate recognition data files")
else:
print("R1. [ ] License plate recognition data files")
if (delete_realtime_object_recognition_data == True):
print("R2. [X] Object recognition data files")
else:
print("R2. [ ] Object recognition data files")
print("")
print("===== Dash-cam Mode =====")
if (delete_dashcam_video == True):
print("D1. [X] Dash-cam videos")
else:
print("D1. [ ] Dash-cam videos")
print("")
selection = prompt("Selection: ", optional=False, input_type=str) # Prompt the user for a selection.
if (selection == "0"):
break
# Toggle the file indicated by the user.
elif (selection.lower() == "m1"):
delete_management_custom = not delete_management_custom
elif (selection.lower() == "p1"):
delete_prerecorded_processed_frames = not delete_prerecorded_processed_frames
elif (selection.lower() == "p2"):
delete_prerecorded_gpx_files = not delete_prerecorded_gpx_files
elif (selection.lower() == "p3"):
delete_prerecorded_license_plate_analysis_data = not delete_prerecorded_license_plate_analysis_data
elif (selection.lower() == "p4"):
delete_prerecorded_object_recognition_data = not delete_prerecorded_object_recognition_data
elif (selection.lower() == "p5"):
delete_prerecorded_license_plate_location_data = not delete_prerecorded_license_plate_location_data
elif (selection.lower() == "r1"):
delete_realtime_license_plate_recognition_data = not delete_realtime_license_plate_recognition_data
elif (selection.lower() == "r2"):
delete_realtime_object_recognition_data = not delete_realtime_object_recognition_data
elif (selection.lower() == "d1"):
delete_dashcam_video = not delete_dashcam_video
if (delete_management_custom):
delete_custom_file_name = prompt("Please specify the name of the additional file you'd like to delete from the current working directory: ")
# Delete the files as per the user's inputs, after confirming the deletion process.
if (prompt("Are you sure you want to delete the selected files permanently? (y/n): ").lower() == "y"):
print("Deleting files...")
if (delete_management_custom):
os.system("rm -r " + config["general"]["working_directory"] + "/" + delete_custom_file_name)
if (delete_prerecorded_processed_frames):
os.system("rm -r " + config["general"]["working_directory"] + "/frames")
if (delete_prerecorded_gpx_files):
os.system("rm " + config["general"]["working_directory"] + "/*.gpx")
if (delete_prerecorded_license_plate_analysis_data):
os.system("rm " + config["general"]["working_directory"] + "/pre_recorded_license_plate_export.*")
if (delete_prerecorded_object_recognition_data):
os.system("rm " + config["general"]["working_directory"] + "/pre_recorded_object_detection_export.*")
if (delete_prerecorded_license_plate_location_data):
os.system("rm " + config["general"]["working_directory"] + "/pre_recorded_location_data_export.*")
if (delete_realtime_license_plate_recognition_data):
os.system("rm " + config["general"]["working_directory"] + "/real_time_plates*")
if (delete_realtime_object_recognition_data):
os.system("rm " + config["general"]["working_directory"] + "/" + config["realtime"]["saving"]["object_recognition"]["file"])
if (delete_dashcam_video):
os.system("rm " + config["general"]["working_directory"] + "/predator_dashcam*")
clear()
print("Files have finished deleting.")
else:
print("No files have been deleted.")
else: # The user has selected an invalid option in the file management menu.
display_message("Invalid selection.", 2)
elif (selection == "2"): # The user has selected the "Information" option.
print(" Please select an option")
print(" 0. Back")
print(" 1. About")
print(" 2. Neofetch")
print(" 3. Print Current Configuration")
if (config["management"]["disk_statistics"] == True): # Check to see if disk statistics are enabled.
print(" 4. Disk Usage") # Display the disk usage option in a normal style.
else: # Otherwise, disk statistics are disabled.
print(" " + style.faint + "4. Disk Usage" + style.end) # Display the disk usage option in a faint style to indicate that it is disabled.
selection = prompt(" Selection: ", optional=False, input_type=str)
if (selection == "0"): # The user has selected to return back to the previous menu.
continue # Do nothing, and just finish this loop.
elif (selection == "1"): # The user has selected the "about" option.
clear()
print(style.bold + "============" + style.end)
print(style.bold + " Predator" + style.end)
print(style.bold + " V0LT" + style.end)
print(style.bold + " V9.0" + style.end)
print(style.bold + " AGPLv3" + style.end)
print(style.bold + "============" + style.end)
elif (selection == "2"): # The user has selected the "neofetch" option.
os.system("neofetch") # Execute neofetch to display information about the system.
elif (selection == "3"): # The user has selected the "print configuration" option.
os.system("cat " + predator_root_directory + "/config.json") # Print out the raw contents of the configuration database.
elif (selection == "4"): # The user has selected the "disk usage" option.
if (config["management"]["disk_statistics"] == True): # Check to make sure disk statistics are enabled before displaying disk statistics.
print("Free space: " + str(round(((psutil.disk_usage(path=config["general"]["working_directory"]).free)/1000000000)*100)/100) + "GB") # Display the free space on the storage device containing the current working directory.
print("Used space: " + str(round(((psutil.disk_usage(path=config["general"]["working_directory"]).used)/1000000000)*100)/100) + "GB") # Display the used space on the storage device containing the current working directory.
print("Total space: " + str(round(((psutil.disk_usage(path=config["general"]["working_directory"]).total)/1000000000)*100)/100) + "GB") # Display the total space on the storage device containing the current working directory.
else: # Disk statistics are disabled, but the user has selected the disk usage option.
display_message("The disk usage could not be displayed because the 'disk_statistics' configuration option is disabled.", 2)
else: # The user has selected an invalid option in the information menu.
display_message("Invalid selection.", 2)
utils.wait_for_input()
elif (selection == "3"): # The user has selected the "Configuration" option.
print(" Please enter the name of a configuration section to edit")
for section in config: # Iterate through each top-level section of the configuration database, and display them all to the user.
if (type(config[section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection1 = prompt("=== Selection (Tier 1): ", optional=True, input_type=str, default="")
if (selection1 in config): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1]: # Iterate through each second-level section of the configuration database, and display them all to the user.
if (type(config[selection1][section]) is dict): # Check to see if the current entry is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection2 = prompt("======= Selection (Tier 2): ", optional=True, input_type=str, default="")
if (selection2 in config[selection1]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2]: # Iterate through each third-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][section]) is dict): # Check to see if the current element is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection3 = prompt("=========== Selection (Tier 3): ", optional=True, input_type=str, default="")
if (selection3 in config[selection1][selection2]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2][selection3]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2][selection3]: # Iterate through each third-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][selection3][section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][selection3][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection4 = prompt("=============== Selection (Tier 4): ", optional=False, input_type=str)
if (selection4 in config[selection1][selection2][selection3]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2][selection3][selection4]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2][selection3][selection4]: # Iterate through each fourth-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][selection3][selection4][section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][selection3][selection4][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection5 = prompt("=================== Selection (Tier 5): ", optional=False, input_type=str)
if (selection5 in config[selection1][selection2][selection3][selection4]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2][selection3][selection4][selection5]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2][selection3][selection4][selection5]: # Iterate through each fifth-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][selection3][selection4][selection5][section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][selection3][selection4][selection5][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection6 = prompt("======================= Selection (Tier 6): ", optional=False, input_type=str)
if (selection6 in config[selection1][selection2][selection3][selection4][selection5]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2][selection3][selection4][selection5][selection6]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2][selection3][selection4][selection5][selection6]: # Iterate through each sixth-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][selection3][selection4][selection5][selection6][section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][selection3][selection4][selection5][selection6][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection7 = prompt("=========================== Selection (Tier 7): ", optional=False, input_type=str)
if (selection7 in config[selection1][selection2][selection3][selection4][selection5][selection6]): # Check to make sure the section entered by the user actually exists in the configuration database.
if (type(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7]) is dict): # Check to make sure the current selection is a dictionary before trying to iterate through it.
for section in config[selection1][selection2][selection3][selection4][selection5][selection6][selection7]: # Iterate through each sixth-level section of the configuration database, and display them all to the user.
if (type(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7][section]) is dict): # Check to see if the current section we're iterating over is a dictionary.
print(" '" + style.bold + str(section) + style.end + "'") # If the entry is a dictionary, display it in bold.
else:
print(" '" + style.italic + str(section) + style.end + "': '" + str(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7][section]) + "'") # If the entry is not a dictionary (meaning it's an actual configuration value), display it in italics.
selection8 = prompt("=============================== Selection (Tier 8): ", optional=False, input_type=str)
else: # If the current selection isn't a dictionary, assume that it's an configuration entry. (Tier 7)
print(" Current Value: " + str(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7]))
config[selection1][selection2][selection3][selection4][selection5][selection6][selection7] = prompt(" New Value (" + str(type(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7])) + "): ", optional=True, input_type=type(config[selection1][selection2][selection3][selection4][selection5][selection6][selection7]), default="")
elif (selection7 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary, assume that it's an configuration entry. (Tier 6)
print(" Current Value: " + str(config[selection1][selection2][selection3][selection4][selection5][selection6]))
config[selection1][selection2][selection3][selection4][selection5][selection6] = prompt(" New Value (" + str(type(config[selection1][selection2][selection3][selection4][selection5][selection6])) + "): ", optional=True, input_type=type(config[selection1][selection2][selection3][selection4][selection5][selection6]), default="")
elif (selection6 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary, assume that it's an configuration entry. (Tier 5)
print(" Current Value: " + str(config[selection1][selection2][selection3][selection4][selection5]))
config[selection1][selection2][selection3][selection4][selection5] = prompt(" New Value (" + str(type(config[selection1][selection2][selection3][selection4][selection5])) + "): ", optional=True, input_type=type(config[selection1][selection2][selection3][selection4][selection5]), default="")
elif (selection5 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary, assume that it's an configuration entry. (Tier 4)
print(" Current Value: " + str(config[selection1][selection2][selection3][selection4]))
config[selection1][selection2][selection3][selection4] = prompt(" New Value (" + str(type(config[selection1][selection2][selection3][selection4])) + "): ", optional=True, input_type=type(config[selection1][selection2][selection3][selection4]), default="")
elif (selection4 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary, assume that it's an configuration entry. (Tier 3)
print(" Current Value: " + str(config[selection1][selection2][selection3]))
config[selection1][selection2][selection3] = prompt(" New Value (" + str(type(config[selection1][selection2][selection3])) + "): ", optional=True, input_type=type(config[selection1][selection2][selection3]), default="")
elif (selection3 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary or list, assume that it's an configuration entry. (Tier 2)
print(" Current Value: " + str(config[selection1][selection2]))
config[selection1][selection2] = prompt(" New Value (" + str(type(config[selection1][selection2])) + "): ", optional=True, input_type=type(config[selection1][selection2]), default="")
elif (selection2 != ""):
display_message("Unknown configuration entry selected.", 3)
else: # If the current selection isn't a dictionary or list, assume that it's an configuration entry. (Tier 1)
print(" Current Value: " + str(config[selection1]))
config[selection1] = prompt(" New Value (" + str(type(config[selection1])) + "): ", optional=True, input_type=type(config[selection1]), default="")
elif (selection1 != ""):
display_message("Unknown configuration entry selected.", 3)
config_file = open(predator_root_directory + "/config.json", "w") # Open the configuration file.
json.dump(config, config_file, indent=4) # Dump the JSON data into the configuration file on the disk.
config_file.close() # Close the configuration file.
config = json.load(open(predator_root_directory + "/config.json")) # Re-load the configuration database from disk.
else: # The user has selected an invalid option in the main management menu.
display_message("Invalid selection.", 2)
# Pre-recorded mode
elif (mode_selection == "1" and config["general"]["modes"]["enabled"]["prerecorded"] == True): # The user has selected to boot into pre-recorded mode.
debug_message("Started pre-recorded mode")
debug_message("Taking user preferences")
working_directory_input = prompt("Working directory (Default " + config["general"]["working_directory"] + "): ", optional=True, input_type=str)
if (working_directory_input == ""): # If the user leaves the
working_directory_input = config["general"]["working_directory"]
while (os.path.exists(working_directory_input) == False): # Run forever until the user enters a working directory that exists.
display_message("The specified working directory doesn't seem to exist.", 2)
working_directory_input = prompt("Working directory (Default " + config["general"]["working_directory"] + "): ", optional=True, input_type=str)
config["general"]["working_directory"] = working_directory_input
video = prompt("Video file name(s): ", optional=False, input_type=str)
framerate = prompt("Frame analysis interval (Default '1.0'): ", optional=True, input_type=float, default=1.0)
current_formats = ', '.join(config["general"]["alpr"]["validation"]["license_plate_format"])
license_plate_format_input = prompt(f"License plate format, separated by commas (Default '{current_formats}'): ", optional=True, input_type=str)
if (license_plate_format_input == ""): # If the user leaves the license plate format input blank, then use the default.
license_plate_format_input = current_formats
# Convert and store the input string as a list of formats
config["general"]["alpr"]["validation"]["license_plate_format"] = [format.strip() for format in license_plate_format_input.split(',')]
video_start_time = prompt("Video starting time (YYYY-mm-dd HH:MM:SS): ", optional=True, input_type=str) # Ask the user when the video recording started so we can correlate it's frames to a GPX file.
if (video_start_time != ""):
gpx_file = prompt("GPX file name: ", optional=True, input_type=str)
if (gpx_file != ""):
while (os.path.exists(config["general"]["working_directory"] + "/" + gpx_file) == False): # Check to see if the GPX file name supplied by the user actually exists in the working directory.
display_message("The specified GPX file does not appear to exists.", 2)
gpx_file = prompt("GPX file name: ", optional=False, input_type=str)
else:
gpx_file = ""
debug_message("Processing user preferences")
if (video_start_time == ""): # If the video_start_time preference was left blank, then default to 0.
video_start_time = 0
else:
try:
video_start_time = round(time.mktime(datetime.datetime.strptime(video_start_time, "%Y-%m-%d %H:%M:%S").timetuple())) # Convert the video_start_time human readable date and time into a Unix timestamp.
except:
display_message("The video starting time specified doesn't appear to be valid. The starting time has been reset to 0. GPX correlation will almost certainly fail.", 3)
video_start_time = 0
if (video[0] == "*"): # Check to see if the first character is a wildcard.
video_list_command = "ls " + config["general"]["working_directory"] + "/" + video + " | tr '\n' ','";
videos = str(os.popen(video_list_command).read())[:-1].split(",") # Run the command, and record the raw output string.
for key, video in enumerate(videos):
videos[key] = os.path.basename(video)
else:
videos = video.split(", ") # Split the video input into a list, based on the position of commas.
for video in videos: # Iterate through each video specified by the user.
if (os.path.exists(config["general"]["working_directory"] + "/" + video) == False): # Check to see if each video file name supplied by the user actually exists in the working directory.
display_message("The video file " + str(video) + " entered doesn't seem to exist in the working directory. Predator will almost certainly fail.", 3) # Inform the user that this video file couldn't be found.
clear() # Clear the console output
# Split the supplied video(s) into individual frames based on the user's input.
debug_message("Splitting video into discrete frames")
video_counter = 0 # Create a placeholder counter that will be incremented by 1 for each video. This will be appended to the file names of the video frames to keep frames from different videos separate.
print("Splitting video into discrete images...")
if (os.path.exists(config["general"]["working_directory"] + "/frames/")): # Check to see the frames directory already exists.
os.system("rm -r " + config["general"]["working_directory"] + "/frames") # Remove the frames directory.
for video in videos: # Iterate through each video specified by the user.
video_counter+=1 # Increment the video counter by 1.
frame_split_command = "mkdir " + config["general"]["working_directory"] + "/frames; ffmpeg -i " + config["general"]["working_directory"] + "/" + video + " -r " + str(1/framerate) + " " + config["general"]["working_directory"] + "/frames/video" + str(video_counter) + "output%04d.png -loglevel quiet" # Set up the FFMPEG command that will be used to split each video into individual frames.
os.system(frame_split_command) # Execute the FFMPEG command to split the video into individual frames.
print("Done.\n")
# Gather all of the individual frames generated previously.
debug_message("Collecting discrete frames")
print("Gathering generated frames...")
frames = os.listdir(config["general"]["working_directory"] + "/frames") # Get all of the files in the folder designated for individual frames.
frames.sort() # Sort the list alphabetically.
print("Done.\n")
# Crop the individual frames to make license plate recognition more efficient and accurate.
if (config["prerecorded"]["image"]["processing"]["cropping"]["enabled"] == True): # Check to see if cropping is enabled in pre-recorded mode.
debug_message("Cropping discrete frames")
print("Cropping individual frames...")
crop_script_path = predator_root_directory + "/crop_image" # Path to the cropping script in the Predator directory.
for frame in frames:
os.system(crop_script_path + " " + config["general"]["working_directory"] + "/frames/" + frame + " " + str(config["prerecorded"]["image"]["processing"]["cropping"]["left_margin"]) + " " + str(config["prerecorded"]["image"]["processing"]["cropping"]["right_margin"]) + " " + str(config["prerecorded"]["image"]["processing"]["cropping"]["top_margin"]) + " " + str(config["prerecorded"]["image"]["processing"]["cropping"]["bottom_margin"]))
print("Done.\n")
# If enabled, count how many objects are in each frame.
if (config["general"]["object_recognition"]["enabled"] == True):
debug_message("Running object recognition")
print("Running object recognition...")
time.sleep(1) # Wait for a short period of time to allow the images to finish saving.
object_count = {} # Create an empty dictionary that will hold each frame and the object recognition counts.
for frame in frames: # Iterate through each frame.
object_count[frame] = {} # Initial the dictionary for this frame.
frame_path = config["general"]["working_directory"] + "/frames/" + frame # Set the file path of the current frame.
image = cv2.imread(frame_path) # Load the frame.
object_recognition_bounding_box, object_recognition_labels, object_recognition_confidence = cv.detect_common_objects(image) # Analyze the image.
for object_recognized in object_recognition_labels: # Iterate through each object recognized.
if (object_recognized in object_count[frame]):
object_count[frame][object_recognized]+=1
else:
object_count[frame][object_recognized] = 1
print("Done.\n")
# Analyze each individual frame, and collect possible plate IDs.
debug_message("Running ALPR")
print("Scanning for license plates...")
alpr_frames = {} # Create an empty dictionary that will hold each frame and the potential license plates IDs.
for frame in frames: # Iterate through each frame of video.
alpr_frames[frame] = {} # Set the license plate recognition information for this frame to an empty list as a placeholder.
# Run license plate analysis on this frame.
reading_output = alpr.run_alpr(config["general"]["working_directory"] + "/frames/" + frame)
# Organize all of the detected license plates and their list of potential guess candidates to a dictionary to make them easier to manipulate.
all_current_plate_guesses = {} # Create an empty place-holder dictionary that will be used to store all of the potential plates and their guesses.
plate_index = 0 # Reset the plate index counter to 0 before the loop.
for detected_plate in reading_output["results"]: # Iterate through each potential plate detected by the ALPR command.
all_current_plate_guesses[plate_index] = {} # Create an empty dictionary for this plate so we can add all the potential plate guesses to it in the next step.
for plate_guess in detected_plate["candidates"]: # Iterate through each plate guess candidate for each potential plate detected.
all_current_plate_guesses[plate_index][plate_guess["plate"]] = plate_guess["confidence"] # Add the current plate guess candidate to the list of plate guesses.
plate_index+=1 # Increment the plate index counter.
if (len(all_current_plate_guesses) > 0): # Only add license plate data to the current frame if data actually exists to add in the first place.
alpr_frames[frame] = all_current_plate_guesses # Record all of the detected plates for this frame.
print("Done.\n")
# Check the possible plate IDs and validate based on general plate formatting specified by the user.
debug_message("Validating ALPR results")
print("Validating license plates...")
validated_alpr_frames = {} # This is a placeholder variable that will be used to store the validated ALPR information for each frame.
# Handle ignore list processing.
for frame in alpr_frames: # Iterate through each frame of video in the database of scanned plates.
validated_alpr_frames[frame] = {} # Set the validated license plate recognition information for this frame to an empty list as a placeholder.
for plate in alpr_frames[frame].keys(): # Iterate through each plate detected per frame.
for guess in alpr_frames[frame][plate]: # Iterate through each guess for each plate.
for ignore_plate in ignore_list: # Iterate through each plate in the ignore list.
if (fnmatch.fnmatch(guess, ignore_plate)): # Check to see if this guess matches a plate in the ignore list.
alpr_frames[frame][plate] = [] # Remove this plate from the ALPR dictionary.
break # Break the loop, since this entire plate, including all of its guesses, has just been removed.
if (fnmatch.fnmatch(guess, config["developer"]["kill_plate"]) and config["developer"]["kill_plate"] != ""): # Check to see if this plate matches the kill plate, and if a kill plate is set.
exit() # Terminate the program.
# Remove any empty plates.
for frame in alpr_frames: # Iterate through each frame of video in the database of scanned plates.
plates = list(alpr_frames[frame].keys())
for plate in plates:
if (len(alpr_frames[frame][plate]) <= 0):
del alpr_frames[frame][plate]
# Handle formatting validation.
for frame in alpr_frames: # Iterate through each frame of video in the database of scanned plates.
validated_alpr_frames[frame] = {} # Set the validated license plate recognition information for this frame to an empty list as a placeholder.
for plate in alpr_frames[frame]: # Iterate through each plate detected per frame.
for guess in alpr_frames[frame][plate]: # Iterate through each guess for each plate.
if (alpr_frames[frame][plate][guess] >= float(config["general"]["alpr"]["validation"]["confidence"])): # Check to make sure this plate's confidence is higher than the minimum threshold set in the configuration.
if any(alpr.validate_plate(guess, format_template) for format_template in config["general"]["alpr"]["validation"]["license_plate_format"]) or "" in config["general"]["alpr"]["validation"]["license_plate_format"]: # Check to see if this plate passes validation.
if (plate not in validated_alpr_frames[frame]): # Check to see if this plate hasn't been added to the validated information yet.
validated_alpr_frames[frame][plate] = [] # Add the plate to the validated information as a blank placeholder list.
validated_alpr_frames[frame][plate].append(guess) # Since this plate guess failed the validation test, delete it from the list of guesses.
print("Done.\n")
# Run through the data for each frame, and save only the first (most likely) license plate to the list of detected plates.
debug_message("Organizing ALPR results")
print("Collecting most likely plate per guess...")
plates_detected = [] # Create an empty list that the detected plates will be added to.
for frame in validated_alpr_frames: # Iterate through all frames.
for plate in validated_alpr_frames[frame]: # Iterate through all plates detected this frame.
if (len(validated_alpr_frames[frame][plate]) > 0): # Check to see if this plate has any guesses associated with it.
plates_detected.append(validated_alpr_frames[frame][plate][0]) # Add the first guess for this plate to this list of plates detected.
print("Done.\n")
# De-duplicate the list of license plates detected.
print("De-duplicating detected license plates...")
plates_detected = list(dict.fromkeys(plates_detected))
print("Done.\n")
debug_message("Checking for alerts")
print("Checking for alerts...")
alert_database = load_alert_database(config["general"]["alerts"]["databases"], config["general"]["working_directory"]) # Load the license plate alert database.
active_alerts = {} # This is an empty placeholder that will hold all of the active alerts.
if (len(alert_database) > 0): # Only run alert processing if the alert database isn't empty.
for rule in alert_database: # Run through every plate in the alert plate database supplied by the user.
for frame in alpr_frames: # Iterate through each frame of video in the raw ALPR data.
if (config["general"]["alerts"]["alerts_ignore_validation"] == True): # If the user has enabled alerts that ignore license plate validation, then use the unvalidated ALPR information.
alpr_frames_to_scan = alpr_frames
else: # If the user hasn't enabled alerts that ignore license plate validation, then use the validated ALPR information.
alpr_frames_to_scan = validated_alpr_frames
for plate in alpr_frames_to_scan[frame]: # Iterate through each of the plates detected this round, regardless of whether or not they were validated.
for guess in alpr_frames_to_scan[frame][plate]: # Run through each of the plate guesses generated by ALPR, regardless of whether or not they are valid according to the plate formatting guideline.
if (fnmatch.fnmatch(guess, rule)): # Check to see this detected plate guess matches this particular plate in the alert database, taking wildcards into account.
active_alerts[guess] = alert_database[rule] # Add this plate to the active alerts dictionary.
active_alerts[guess]["rule"] = rule # Add the rule that triggered this alert to the alert information.
active_alerts[guess]["frame"] = frame # Add the rule that triggered this alert to the alert information.
if (config["general"]["alerts"]["allow_duplicate_alerts"] == False):
break # Break the loop if an alert is found for this guess, in order to avoid triggering multiple alerts for each guess of the same plate.
display_alerts(active_alerts) # Display all active alerts.
print("Done.\n")
# Correlate the detected license plates with a GPX file.
frame_locations = {} # Create a blank database that will be used during the process
if (gpx_file != ""): # Check to make sure the user actually supplied a GPX file.
debug_message("Correlated location data")
print("Processing location data...")
decoded_gpx_data = process_gpx(config["general"]["working_directory"] + "/" + gpx_file) # Decode the data from the GPX file.
iteration = 0 # Set the iteration counter to 0 so we can add one to it each frame we iterate through.
for element in alpr_frames: # Iterate through each frame.
iteration+=1 # Add one to the iteration counter.