423 lines
16 KiB
Python
423 lines
16 KiB
Python
import os
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
|
|
import serial
|
|
import serial.tools.list_ports
|
|
import tkinter as tk
|
|
from tkinter import filedialog, messagebox, ttk
|
|
|
|
from . import globals as g
|
|
from .frm_guns import FrmGuns
|
|
from .gun_object import GunObject
|
|
|
|
|
|
def bytes_to_hex(data):
|
|
return "".join("{:02X}".format(b) for b in data)
|
|
|
|
|
|
class MainWindow:
|
|
|
|
def __init__(self, root):
|
|
self.root = root
|
|
self.root.title("LBMS Console v1")
|
|
self.root.geometry("1408x863")
|
|
self.font = ("Consolas", 10)
|
|
|
|
self.save_data = datetime.now() + timedelta(seconds=60)
|
|
self.valid_counter = 0
|
|
self.limit_fire = 11000
|
|
self.limit_battery = 660
|
|
self.gun_objects = {}
|
|
self.serial_port = None
|
|
self.sim_list = []
|
|
self.time_offset = 0
|
|
self.list_index = 0
|
|
self.simulate_enabled = False
|
|
self.show_enabled = False
|
|
self._reader_thread = None
|
|
self._reader_stop = threading.Event()
|
|
|
|
self.button1 = tk.Button(self.root, text="Button1", font=self.font, command=self._button1_click)
|
|
self.button1.place(x=0, y=0, width=87, height=27)
|
|
self.button2 = tk.Button(self.root, text="Button2", font=self.font, command=self._button2_click)
|
|
self.button2.place(x=90, y=0, width=87, height=27)
|
|
self.txt_comm_port = ttk.Combobox(self.root, font=self.font,
|
|
values=[p.device for p in serial.tools.list_ports.comports()])
|
|
self.txt_comm_port.set("COM4")
|
|
self.txt_comm_port.place(x=112, y=3, width=116)
|
|
self.label1 = tk.Label(self.root, text="Label1", font=self.font)
|
|
self.label1.place(x=247, y=12)
|
|
self.label2 = tk.Label(self.root, text="Label2", font=self.font)
|
|
self.label2.place(x=12, y=30)
|
|
self.btn_simulate = tk.Button(self.root, text="Simulate", font=self.font, command=self._btn_simulate_click)
|
|
self.btn_simulate.place(x=405, y=2, width=75, height=23)
|
|
self.label3 = tk.Label(self.root, text="Label3", font=self.font)
|
|
self.label3.place(x=505, y=3)
|
|
self.label4 = tk.Label(self.root, text="Label4", font=self.font)
|
|
self.label4.place(x=505, y=18)
|
|
|
|
self.text_box2 = tk.Text(self.root, bg="black", fg="#00c000", font=self.font)
|
|
self.text_box2.place(x=14, y=48, width=477, height=784)
|
|
self.text_box3 = tk.Text(self.root, bg="black", fg="#00c000", font=("Courier", 10, "bold"))
|
|
self.text_box3.place(x=497, y=48, width=899, height=784)
|
|
|
|
self.root.bind("<Configure>", self._on_resize)
|
|
self.root.protocol("WM_DELETE_WINDOW", self._on_closing)
|
|
|
|
self._timer_loop(75, self._timer1_tick)
|
|
self._timer_loop(5, self._tmr_simulate_tick)
|
|
self._timer_loop(1000, self._tmr_show_tick)
|
|
|
|
self._on_load()
|
|
|
|
def _timer_loop(self, ms, func):
|
|
def loop():
|
|
try:
|
|
func()
|
|
finally:
|
|
self.root.after(ms, loop)
|
|
loop()
|
|
|
|
def _on_resize(self, event):
|
|
if event.widget is not self.root:
|
|
return
|
|
width = event.width
|
|
height = event.height
|
|
self.text_box2.place(x=14, y=48, width=width // 2 - 25, height=height - 48)
|
|
self.text_box3.place(x=width // 2 + 11, y=48, width=width - width // 2 - 23, height=height - 48)
|
|
|
|
def write_error(self, str_msg):
|
|
with open("err.log", "a", encoding="utf-8") as f:
|
|
f.write(str_msg + "\n")
|
|
|
|
def _fatal(self, str_msg):
|
|
self.write_error(str_msg)
|
|
os._exit(1)
|
|
|
|
def _on_load(self):
|
|
arg_com = ""
|
|
for arg in sys.argv[1:]:
|
|
if "=" not in arg:
|
|
continue
|
|
key, _, value = arg.partition("=")
|
|
key = key.lower()
|
|
if key == "admin":
|
|
if value.lower() == "true":
|
|
g.admin_mode = True
|
|
elif key == "com":
|
|
arg_com = value.strip()
|
|
elif key == "fired":
|
|
self.limit_fire = int(value.strip())
|
|
elif key == "battery":
|
|
self.limit_battery = int(value.strip())
|
|
|
|
self.save_data = datetime.now() + timedelta(seconds=60)
|
|
self.serial_port = serial.Serial(baudrate=115200, bytesize=serial.EIGHTBITS,
|
|
parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE)
|
|
g.message_queue = queue.Queue()
|
|
g.message_dequeue = ""
|
|
self._create_id_list()
|
|
self.gun_objects = {}
|
|
self.frm_guns = FrmGuns(self.root, self)
|
|
self.frm_guns.lift()
|
|
|
|
items_dir = "./items"
|
|
if os.path.isdir(items_dir):
|
|
for file_name in sorted(os.listdir(items_dir)):
|
|
if file_name.endswith(".cfg"):
|
|
t_go = GunObject.load_gun_info(os.path.join(items_dir, file_name))
|
|
self.gun_objects[t_go.gun_mac] = t_go
|
|
else:
|
|
for mac, gun_id in g.gun_list.items():
|
|
self.gun_objects[mac] = GunObject(gun_id, mac)
|
|
|
|
for item in self.gun_objects.values():
|
|
item.settings_battery_life = self.limit_battery
|
|
item.settings_fired_max = self.limit_fire
|
|
self.frm_guns.add_gun_button(item)
|
|
self.frm_guns.align_icons()
|
|
self.show_enabled = True
|
|
|
|
if arg_com != "":
|
|
self._auto_start(arg_com)
|
|
self.root.iconify()
|
|
|
|
def _create_id_list(self):
|
|
g.gun_list = {
|
|
"0013A20040A04224": "01",
|
|
"0013A20040A0422D": "02",
|
|
"0013A200408BD898": "03",
|
|
"0013A20040A46DD3": "04",
|
|
"0013A20040A043AE": "05",
|
|
"0013A20040A042AB": "06",
|
|
"0013A20040A041EF": "07",
|
|
"0013A20040A041D4": "08",
|
|
"0013A20040A0430E": "09",
|
|
"0013A20040A043CD": "10",
|
|
"0013A20040A042D0": "11",
|
|
"0013A20040A46E00": "12",
|
|
"0013A20040A043B1": "13",
|
|
"0013A200409F0D5A": "14",
|
|
"0013A20040A04430": "15",
|
|
"0013A20040A04394": "16",
|
|
"0013A20040A043E1": "17",
|
|
"0013A20040A0439B": "18",
|
|
"0013A200409B772B": "19",
|
|
"0013A2004093E3CB": "20",
|
|
"0013A200409B7B9A": "21",
|
|
"0013A2004093E3D5": "22",
|
|
"0013A20040A779AA": "23",
|
|
"0013A200409C9FCA": "24",
|
|
"0013A20040AF81C8": "25",
|
|
"0013A20040B1C483": "26",
|
|
"0013A20040B1C440": "27",
|
|
"0013A200409C9FD0": "28",
|
|
"0013A200409C9FF8": "29",
|
|
"0013A200409E4383": "30",
|
|
"0013A200409CA02A": "31",
|
|
"0013A200409C9FEC": "32",
|
|
"0013A200409E438B": "33",
|
|
"0013A20040E5C3D1": "34",
|
|
"0013A20040F303F6": "35",
|
|
"0013A200409E4352": "36",
|
|
"0013A200409CA071": "37",
|
|
"0013A20040A779BB": "38",
|
|
"0013A20040B17596": "39",
|
|
"0013A20040F303E1": "40",
|
|
"0013A20040F149C4": "41",
|
|
"0013A20040E5C66C": "42",
|
|
"0013A20040D947EE": "43",
|
|
"0013A20040B17EB2": "44",
|
|
}
|
|
|
|
def get_gun_id(self, gun_id):
|
|
return g.gun_list.get(gun_id, gun_id)
|
|
|
|
def _auto_start(self, com_port):
|
|
try:
|
|
self.serial_port.port = com_port
|
|
self.serial_port.open()
|
|
self._start_reader()
|
|
except Exception as ex:
|
|
messagebox.showerror("ERROR!!!", str(ex))
|
|
|
|
def _button1_click(self):
|
|
if self.serial_port is not None and self.serial_port.is_open:
|
|
self._close_serial()
|
|
else:
|
|
self.serial_port.port = self.txt_comm_port.get()
|
|
self.serial_port.open()
|
|
self._start_reader()
|
|
|
|
def _close_serial(self):
|
|
self._reader_stop.set()
|
|
if self.serial_port.is_open:
|
|
self.serial_port.close()
|
|
|
|
def _start_reader(self):
|
|
self._reader_stop.clear()
|
|
if self._reader_thread is not None and self._reader_thread.is_alive():
|
|
return
|
|
self._reader_thread = threading.Thread(target=self._serial_reader, daemon=True)
|
|
self._reader_thread.start()
|
|
|
|
def _serial_reader(self):
|
|
while not self._reader_stop.is_set():
|
|
try:
|
|
if not self.serial_port.is_open:
|
|
break
|
|
bytes_to_read = self.serial_port.in_waiting
|
|
if bytes_to_read > 0:
|
|
num_array = self.serial_port.read(bytes_to_read)
|
|
g.message_queue.put(bytes_to_hex(num_array))
|
|
except Exception as ex:
|
|
self.write_error("read: " + str(ex))
|
|
break
|
|
time.sleep(0.005)
|
|
|
|
def _timer1_tick(self):
|
|
now = datetime.now()
|
|
try:
|
|
if self.save_data < now:
|
|
self.save_data = now + timedelta(seconds=61)
|
|
self.save_guns()
|
|
if self.serial_port is not None and self.serial_port.is_open:
|
|
self.label1.configure(text="Open")
|
|
else:
|
|
self.label1.configure(text="Closed")
|
|
self.label2.configure(text=str(g.message_queue.qsize()))
|
|
except Exception as ex:
|
|
self._fatal("1" + str(ex))
|
|
return
|
|
tmp_str = ""
|
|
try:
|
|
q_amount = g.message_queue.qsize()
|
|
if q_amount > 6:
|
|
count_t = int(q_amount * 0.5)
|
|
for _ in range(count_t):
|
|
tmp_str += g.message_queue.get()
|
|
elif 1 <= q_amount <= 6:
|
|
tmp_str = g.message_queue.get()
|
|
g.message_dequeue += tmp_str
|
|
except Exception as ex:
|
|
self._fatal("2" + str(ex))
|
|
return
|
|
try:
|
|
self._dequeue_message()
|
|
for _ in range(11):
|
|
if len(g.message_dequeue) > 200:
|
|
self._dequeue_message()
|
|
except Exception as ex:
|
|
self._fatal("3" + str(ex))
|
|
return
|
|
try:
|
|
if len(g.message_dequeue) > 16:
|
|
c_count = 0
|
|
for x in range(6, len(g.message_dequeue) - 2, 2):
|
|
c_count += int(g.message_dequeue[x:x + 2], 16)
|
|
stats = ("Lines Read : {}\n"
|
|
"Valid Packets: {}\n"
|
|
"Queue Lenght: {}\n"
|
|
"Decode Lenght: {}\n"
|
|
"Packet Lenght: {}\n"
|
|
"\n"
|
|
"\n"
|
|
"{}").format(self.list_index,
|
|
self.valid_counter,
|
|
g.message_queue.qsize(),
|
|
len(g.message_dequeue),
|
|
len(tmp_str),
|
|
g.message_dequeue)
|
|
self.text_box2.delete("1.0", "end")
|
|
self.text_box2.insert("1.0", stats)
|
|
except Exception as ex:
|
|
self._fatal("4" + str(ex))
|
|
|
|
def _dequeue_message(self):
|
|
valid_message = False
|
|
try:
|
|
if len(g.message_dequeue) >= 92:
|
|
tmp_start = g.message_dequeue.find("7E002A80")
|
|
if tmp_start >= 0:
|
|
if len(g.message_dequeue) >= tmp_start + 80:
|
|
x_addr = g.message_dequeue[tmp_start + 8:tmp_start + 24]
|
|
g.message_dequeue = g.message_dequeue[tmp_start + 92:]
|
|
self.valid_counter += 1
|
|
self._add_seen(x_addr)
|
|
valid_message = True
|
|
if len(g.message_dequeue) >= 80:
|
|
tmp_start = g.message_dequeue.find("7E002480")
|
|
if tmp_start >= 0:
|
|
if len(g.message_dequeue) >= tmp_start + 80:
|
|
x_addr = g.message_dequeue[tmp_start + 8:tmp_start + 24]
|
|
x_fire = g.message_dequeue[tmp_start + 8 + 16 + 34:tmp_start + 8 + 16 + 34 + 4]
|
|
g.message_dequeue = g.message_dequeue[tmp_start + 8 + 16 + 34 + 4 + 18:]
|
|
self.valid_counter += 1
|
|
tmpfire1 = int(x_fire[0:2], 16)
|
|
tmpfire2 = int(x_fire[2:4], 16)
|
|
tmp_fire = (tmpfire1 * 256) + tmpfire2
|
|
self._add_count(x_addr, tmp_fire)
|
|
self._add_seen(x_addr)
|
|
valid_message = True
|
|
if len(g.message_dequeue) >= 60:
|
|
tmp_start = g.message_dequeue.find("7E002480")
|
|
if tmp_start >= 0:
|
|
if len(g.message_dequeue) >= tmp_start + 80:
|
|
x_addr = g.message_dequeue[tmp_start + 8:tmp_start + 24]
|
|
g.message_dequeue = g.message_dequeue[tmp_start + 8 + 16 + 36:]
|
|
self.valid_counter += 1
|
|
self._add_seen(x_addr)
|
|
valid_message = True
|
|
if not valid_message and len(g.message_dequeue) >= 200:
|
|
g.message_dequeue = g.message_dequeue[len(g.message_dequeue) // 2:]
|
|
except Exception:
|
|
pass
|
|
|
|
def _add_seen(self, mac):
|
|
gun = self.gun_objects.get(mac)
|
|
if gun is None:
|
|
return
|
|
now = datetime.now()
|
|
if gun.last_count < now - timedelta(minutes=g.count_interval):
|
|
gun.battery += g.count_interval
|
|
gun.last_count = now
|
|
gun.last_seen = now
|
|
|
|
def _add_count(self, mac, count):
|
|
gun = self.gun_objects.get(mac)
|
|
if gun is None:
|
|
return
|
|
if count < gun.fired:
|
|
gun.fired_overall += gun.fired
|
|
gun.battery_overall += gun.battery
|
|
gun.battery = 0
|
|
elif count > gun.fired:
|
|
gun.fired_last = datetime.now() + timedelta(seconds=3)
|
|
gun.fired = count
|
|
|
|
def _btn_simulate_click(self):
|
|
if self.simulate_enabled:
|
|
self.simulate_enabled = False
|
|
self.label4.configure(text=str(datetime.now()))
|
|
self.btn_simulate.configure(text="Simulate")
|
|
return
|
|
self.sim_list = []
|
|
g.message_dequeue = ""
|
|
the_file = filedialog.askopenfilename()
|
|
if not the_file:
|
|
return
|
|
self.list_index = 0
|
|
with open(the_file, "r", encoding="utf-8", errors="replace") as file_reader:
|
|
for line in file_reader:
|
|
self.sim_list.append(line.rstrip("\r\n"))
|
|
|
|
first_ts = datetime.fromisoformat(self.sim_list[0].split(",")[0])
|
|
self.time_offset = (first_ts - datetime.now()).total_seconds()
|
|
|
|
self.label3.configure(text=str(datetime.now()))
|
|
self.label4.configure(text="")
|
|
self.btn_simulate.configure(text="Stop")
|
|
self.simulate_enabled = True
|
|
|
|
def _tmr_simulate_tick(self):
|
|
if not self.simulate_enabled:
|
|
return
|
|
if self.list_index >= len(self.sim_list) - 1:
|
|
self.simulate_enabled = False
|
|
self.btn_simulate.configure(text="Simulate.")
|
|
self.label4.configure(text=str(datetime.now()))
|
|
return
|
|
parts = self.sim_list[self.list_index].split(",")
|
|
if datetime.now() + timedelta(seconds=self.time_offset) > datetime.fromisoformat(parts[0]):
|
|
g.message_queue.put(parts[1])
|
|
self.list_index += 1
|
|
|
|
def _tmr_show_tick(self):
|
|
if not self.show_enabled:
|
|
return
|
|
parts = []
|
|
for item in self.gun_objects.values():
|
|
parts.append("{} {:04d}-{:04d} ({:04d})".format(item.gun_id,
|
|
item.fired,
|
|
item.fired_overall,
|
|
item.fired + item.fired_overall))
|
|
parts.append(" T: {:03d} ({})\n".format(item.battery + item.battery_overall, item.last_seen))
|
|
self.text_box3.delete("1.0", "end")
|
|
self.text_box3.insert("1.0", "".join(parts))
|
|
|
|
def _button2_click(self):
|
|
self.save_guns()
|
|
messagebox.showinfo("LBMS", "done")
|
|
|
|
def save_guns(self):
|
|
for item in self.gun_objects.values():
|
|
item.store_gun_info()
|
|
|
|
def _on_closing(self):
|
|
self.save_guns()
|
|
self.root.destroy()
|