370 lines
15 KiB
Python
370 lines
15 KiB
Python
import tkinter as tk
|
|
from tkinter import filedialog, messagebox, ttk
|
|
import threading
|
|
import socket
|
|
import time
|
|
import binascii
|
|
import requests
|
|
import json # To handle structured data better
|
|
|
|
def hexspace(data, n):
|
|
return " ".join([data[i:i + n] for i in range(0, len(data), n)])
|
|
|
|
session = requests.Session()
|
|
session.headers.update({"Connection": "keep-alive"})
|
|
|
|
class DataReceiver:
|
|
def __init__(self):
|
|
self.running = False
|
|
self.file_mode = False
|
|
self.file_path = None
|
|
self.manual_mode = False
|
|
|
|
def start_tcp(self, host, port, callback):
|
|
self.running = True
|
|
def run():
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind((host, port))
|
|
s.listen(1)
|
|
conn, addr = s.accept()
|
|
with conn:
|
|
while self.running:
|
|
data = conn.recv(1024)
|
|
if data:
|
|
callback(data.decode('utf-8'))
|
|
threading.Thread(target=run, daemon=True).start()
|
|
|
|
def start_file(self, callback, timeout):
|
|
self.running = True
|
|
def run():
|
|
if not self.file_path:
|
|
return
|
|
with open(self.file_path, 'r') as f:
|
|
while self.running:
|
|
if self.manual_mode:
|
|
continue
|
|
line = f.readline()
|
|
if line:
|
|
try:
|
|
# Safely parse the data, assuming the data format is correct
|
|
callback(json.loads(line.strip().split(" DEBUG ")[1]))
|
|
except Exception as e:
|
|
print(f"Error parsing line: {e}")
|
|
time.sleep(timeout)
|
|
threading.Thread(target=run, daemon=True).start()
|
|
|
|
def read_next_line(self, callback):
|
|
if not self.file_path:
|
|
return
|
|
with open(self.file_path, 'r') as f:
|
|
line = f.readline()
|
|
if line:
|
|
try:
|
|
callback(json.loads(line.strip().split(" DEBUG ")[1]))
|
|
except Exception as e:
|
|
print(f"Error parsing line: {e}")
|
|
|
|
def stop(self):
|
|
self.running = False
|
|
|
|
class DataSender:
|
|
def __init__(self):
|
|
self.targets = []
|
|
|
|
def add_target(self, host, port):
|
|
self.targets.append((host, port))
|
|
|
|
def send(self, data):
|
|
for host, port in self.targets:
|
|
try:
|
|
url = f"http://{host}:{port}/API/"
|
|
for k, v in data.items():
|
|
par = "SetText" if k.split(".")[1] == "Text" else "SetImage"
|
|
params = {
|
|
"Function": par,
|
|
"Input": 23,
|
|
"SelectedName": k,
|
|
"Value": v,
|
|
}
|
|
session.get(url, params=params)
|
|
except Exception as e:
|
|
print(f"Error sending data to {host}:{port}: {e}")
|
|
|
|
class DataProcessor:
|
|
def process_protocol_a(self, data):
|
|
return data.upper()
|
|
|
|
def process_protocol_b(self, data):
|
|
return data[::-1]
|
|
|
|
def process_protocol_saratov(self, data):
|
|
if data == "\xff":
|
|
return None
|
|
|
|
cdata = binascii.hexlify(data.encode()) # Ensure data is properly encoded to bytes
|
|
ddata = cdata.decode("utf-8").upper()
|
|
edata = hexspace(ddata, 2)
|
|
temp = edata.split("FF 52")
|
|
processed_data = []
|
|
for i in temp:
|
|
try:
|
|
parts = i.split()
|
|
minutes = int(parts[5], 16)
|
|
seconds = int(parts[6], 16)
|
|
timer_str = f"{minutes}:{seconds:02d}" if seconds < 60 else f"{minutes}.{seconds - 128}"
|
|
|
|
seconds_24 = int(parts[7], 16)
|
|
timer_attack = ""
|
|
timer_pic = "D:/Графика/БАСКЕТБОЛ/ЕДИНАЯ ЛИГА ВТБ 2022-2023/Scorebug Indicators/24Sec_Orange.png"
|
|
if seconds_24 != 255:
|
|
if seconds_24 > 127:
|
|
seconds_24 -= 128
|
|
timer_attack = f"{seconds_24 // 10}.{seconds_24 % 10}"
|
|
timer_pic = "D:/Графика/БАСКЕТБОЛ/ЕДИНАЯ ЛИГА ВТБ 2022-2023/Scorebug Indicators/24Sec_Red.png"
|
|
else:
|
|
timer_attack = seconds_24
|
|
|
|
quarter = int(parts[14][1], 16)
|
|
timeout1 = int(parts[10], 16)
|
|
timeout2 = int(parts[11], 16)
|
|
score1 = int(parts[8], 16)
|
|
score2 = int(parts[9], 16)
|
|
foul1 = int(parts[12], 16)
|
|
foul2 = int(parts[13], 16)
|
|
|
|
data_dict = {
|
|
"TIMER.Text": timer_str,
|
|
"ATTACK.Text": timer_attack,
|
|
"Score_Home.Text": score1,
|
|
"Score_Away.Text": score2,
|
|
"fouls1.Source": f"D:/Графика/БАСКЕТБОЛ/ЕДИНАЯ ЛИГА ВТБ 2022-2023/Scorebug Indicators/Home_{foul1}.png",
|
|
"fouls2.Source": f"D:/Графика/БАСКЕТБОЛ/ЕДИНАЯ ЛИГА ВТБ 2022-2023/Scorebug Indicators/Away_{foul2}.png",
|
|
"24SECBACKGROUND.Source": timer_pic,
|
|
}
|
|
|
|
processed_data.append(data_dict)
|
|
except (IndexError, ValueError) as e:
|
|
print(f"Error processing data: {e}")
|
|
continue
|
|
|
|
return processed_data
|
|
|
|
class App:
|
|
def __init__(self, root):
|
|
self.receiver = DataReceiver()
|
|
self.sender = DataSender()
|
|
self.processor = DataProcessor()
|
|
|
|
self.protocol_var = tk.StringVar(value="Protocol A")
|
|
self.mode_var = tk.StringVar(value="Log File")
|
|
self.read_mode_var = tk.StringVar(value="Automatic")
|
|
self.timeout_var = tk.DoubleVar(value=0.5)
|
|
|
|
self.create_ui(root)
|
|
|
|
def display_data(self, data):
|
|
protocol = self.protocol_var.get()
|
|
if protocol == "Protocol A":
|
|
processed_data = self.processor.process_protocol_a(data)
|
|
elif protocol == "Protocol B":
|
|
processed_data = self.processor.process_protocol_b(data)
|
|
elif protocol == "Saratov":
|
|
processed_data = self.processor.process_protocol_saratov(data)
|
|
else:
|
|
processed_data = data
|
|
|
|
if processed_data:
|
|
self.update_send_status(True)
|
|
self.sender.send(processed_data)
|
|
self.update_send_status(False)
|
|
|
|
self.log_text.insert(tk.END, f"{data}\n")
|
|
self.log_text.see(tk.END)
|
|
|
|
self.processed_data_text.insert(tk.END, f"{processed_data}\n")
|
|
self.processed_data_text.see(tk.END)
|
|
|
|
def create_ui(self, root):
|
|
root.title("Data Processing Application")
|
|
|
|
self.notebook = ttk.Notebook(root)
|
|
self.notebook.pack(fill="both", expand=True)
|
|
|
|
self.settings_frame = tk.Frame(self.notebook)
|
|
self.game_frame = tk.Frame(self.notebook)
|
|
|
|
self.notebook.add(self.settings_frame, text="Settings")
|
|
self.notebook.add(self.game_frame, text="Game")
|
|
|
|
self.create_settings_ui(self.settings_frame)
|
|
self.create_game_ui(self.game_frame)
|
|
|
|
def create_settings_ui(self, frame):
|
|
frame_receive = tk.LabelFrame(frame, text="Receive Data")
|
|
frame_receive.pack(fill="x", padx=5, pady=5)
|
|
|
|
tk.Label(frame_receive, text="Select Mode:").pack(side="left", padx=5, pady=5)
|
|
tk.OptionMenu(frame_receive, self.mode_var, "Log File", "TCP Connection", command=self.update_mode).pack(side="left", padx=5, pady=5)
|
|
|
|
self.file_button = tk.Button(frame_receive, text="Select Log File", command=self.select_file)
|
|
|
|
self.tcp_frame = tk.Frame(frame_receive)
|
|
tk.Label(self.tcp_frame, text="IP:").pack(side="left")
|
|
self.tcp_ip_entry = tk.Entry(self.tcp_frame)
|
|
self.tcp_ip_entry.pack(side="left", padx=5)
|
|
self.tcp_ip_entry.insert(0, "192.168.127.254") # Example default IP address
|
|
tk.Label(self.tcp_frame, text="Port:").pack(side="left")
|
|
self.tcp_port_entry = tk.Entry(self.tcp_frame)
|
|
self.tcp_port_entry.pack(side="left", padx=5)
|
|
self.tcp_port_entry.insert(0, 1993) # Example default port
|
|
|
|
self.read_mode_frame = tk.Frame(frame_receive)
|
|
tk.Label(self.read_mode_frame, text="Read Mode:").pack(side="left", padx=5, pady=5)
|
|
tk.OptionMenu(self.read_mode_frame, self.read_mode_var, "Automatic", "Manual", command=self.update_read_mode).pack(side="left", padx=5, pady=5)
|
|
tk.Label(self.read_mode_frame, text="Timeout (s):").pack(side="left", padx=5, pady=5)
|
|
self.timeout_entry = tk.Entry(self.read_mode_frame, textvariable=self.timeout_var)
|
|
self.timeout_entry.pack(side="left", padx=5)
|
|
self.manual_read_button = tk.Button(self.read_mode_frame, text="Read Next Line", command=self.read_next_line)
|
|
|
|
self.start_button = tk.Button(frame_receive, text="Start", command=self.start_receiving)
|
|
self.start_button.pack(side="left", padx=5, pady=5)
|
|
tk.Button(frame_receive, text="Stop", command=self.stop_receiving).pack(side="left", padx=5, pady=5)
|
|
|
|
frame_process = tk.LabelFrame(frame, text="Process Data")
|
|
frame_process.pack(fill="x", padx=5, pady=5)
|
|
|
|
tk.Label(frame_process, text="Select Protocol:").pack(side="left", padx=5, pady=5)
|
|
tk.OptionMenu(frame_process, self.protocol_var, "Protocol A", "Protocol B", "Saratov").pack(side="left", padx=5, pady=5)
|
|
|
|
frame_send = tk.LabelFrame(frame, text="Send Data")
|
|
frame_send.pack(fill="x", padx=5, pady=5)
|
|
|
|
self.target_count_var = tk.IntVar(value=1)
|
|
tk.Label(frame_send, text="Number of Targets:").pack(side="left", padx=5, pady=5)
|
|
tk.Spinbox(frame_send, from_=1, to=10, textvariable=self.target_count_var, command=self.update_targets).pack(side="left", padx=5, pady=5)
|
|
|
|
self.targets_frame = tk.Frame(frame_send)
|
|
self.targets_frame.pack(fill="x", padx=5, pady=5)
|
|
self.target_entries = []
|
|
self.update_targets()
|
|
|
|
tk.Button(frame_send, text="Send Data", command=self.send_data).pack(side="left", padx=5, pady=5)
|
|
|
|
self.update_mode(self.mode_var.get())
|
|
|
|
def create_game_ui(self, frame):
|
|
frame_status = tk.LabelFrame(frame, text="Status")
|
|
frame_status.pack(fill="x", padx=5, pady=5)
|
|
|
|
self.data_receive_status = tk.Label(frame_status, text="Receiving Data: Not Active", fg="red")
|
|
self.data_receive_status.pack(pady=5)
|
|
|
|
self.data_send_status = tk.Label(frame_status, text="Sending Data: Not Active", fg="red")
|
|
self.data_send_status.pack(pady=5)
|
|
|
|
frame_log = tk.LabelFrame(frame, text="Log")
|
|
frame_log.pack(fill="both", expand=True, padx=5, pady=5)
|
|
|
|
self.log_text = tk.Text(frame_log, height=10, state="normal")
|
|
self.log_text.pack(fill="both", expand=True, padx=5, pady=5)
|
|
|
|
frame_processed = tk.LabelFrame(frame, text="Processed Data")
|
|
frame_processed.pack(fill="both", expand=True, padx=5, pady=5)
|
|
|
|
self.processed_data_text = tk.Text(frame_processed, height=10, state="normal")
|
|
self.processed_data_text.pack(fill="both", expand=True, padx=5, pady=5)
|
|
|
|
def update_mode(self, mode):
|
|
if mode == "Log File":
|
|
self.file_button.pack(side="left", padx=5, pady=5)
|
|
self.tcp_frame.pack_forget()
|
|
self.read_mode_frame.pack(fill="x", padx=5, pady=5)
|
|
self.start_button.config(state="normal")
|
|
else:
|
|
self.file_button.pack_forget()
|
|
self.read_mode_frame.pack_forget()
|
|
self.tcp_frame.pack(side="left", padx=5, pady=5)
|
|
self.start_button.config(state="normal")
|
|
|
|
def update_read_mode(self, mode):
|
|
if mode == "Automatic":
|
|
self.timeout_entry.config(state="normal")
|
|
self.manual_read_button.pack_forget()
|
|
self.receiver.manual_mode = False
|
|
else:
|
|
self.timeout_entry.config(state="disabled")
|
|
self.manual_read_button.pack(side="left", padx=5, pady=5)
|
|
self.receiver.manual_mode = True
|
|
|
|
def select_file(self):
|
|
self.receiver.file_path = filedialog.askopenfilename()
|
|
|
|
def start_receiving(self):
|
|
self.start_button.config(state="disabled")
|
|
mode = self.mode_var.get()
|
|
if mode == "Log File":
|
|
if self.receiver.file_path:
|
|
timeout = self.timeout_var.get() if self.read_mode_var.get() == "Automatic" else 0
|
|
self.receiver.start_file(self.display_data, timeout)
|
|
self.update_receive_status(True)
|
|
else:
|
|
messagebox.showerror("Error", "No file selected.")
|
|
elif mode == "TCP Connection":
|
|
host = self.tcp_ip_entry.get()
|
|
port = self.tcp_port_entry.get()
|
|
if host and port.isdigit():
|
|
self.receiver.start_tcp(host, int(port), self.display_data)
|
|
self.update_receive_status(True)
|
|
else:
|
|
messagebox.showerror("Error", "Invalid IP or port.")
|
|
self.start_button.config(state="normal")
|
|
|
|
def stop_receiving(self):
|
|
self.receiver.stop()
|
|
self.start_button.config(state="normal")
|
|
self.update_receive_status(False)
|
|
|
|
def read_next_line(self):
|
|
self.receiver.read_next_line(self.display_data)
|
|
|
|
def update_receive_status(self, active):
|
|
if active:
|
|
self.data_receive_status.config(text="Receiving Data: Active", fg="green")
|
|
else:
|
|
self.data_receive_status.config(text="Receiving Data: Not Active", fg="red")
|
|
|
|
def update_send_status(self, active):
|
|
if active:
|
|
self.data_send_status.config(text="Sending Data: Active", fg="green")
|
|
else:
|
|
self.data_send_status.config(text="Sending Data: Not Active", fg="red")
|
|
|
|
def update_targets(self):
|
|
for widget in self.targets_frame.winfo_children():
|
|
widget.destroy()
|
|
|
|
self.target_entries = []
|
|
for i in range(self.target_count_var.get()):
|
|
target_frame = tk.Frame(self.targets_frame)
|
|
target_frame.pack(fill="x", pady=2)
|
|
|
|
tk.Label(target_frame, text=f"Target {i+1}:").pack(side="left", padx=5, pady=5)
|
|
host_entry = tk.Entry(target_frame)
|
|
host_entry.pack(side="left", padx=5)
|
|
host_entry.insert(0, f"192.168.1.{i+1}")
|
|
self.target_entries.append(host_entry)
|
|
|
|
def send_data(self):
|
|
if not self.target_entries:
|
|
messagebox.showerror("Error", "No targets defined.")
|
|
return
|
|
|
|
targets = [(entry.get(), 8080) for entry in self.target_entries]
|
|
self.sender.targets = targets
|
|
self.sender.send({"data": "sample"})
|
|
|
|
if __name__ == "__main__":
|
|
root = tk.Tk()
|
|
app = App(root)
|
|
root.mainloop()
|