Compare commits
3 commits
Author | SHA1 | Date | |
---|---|---|---|
ab4f72b91c | |||
3010def921 | |||
ecc1ab1337 |
3 changed files with 200 additions and 6 deletions
176
app.py
Normal file
176
app.py
Normal file
|
@ -0,0 +1,176 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
|
import tkinter as tk
|
||||||
|
from tkinter import ttk, messagebox, simpledialog
|
||||||
|
from typing import final
|
||||||
|
from netbrite import (
|
||||||
|
NetBrite,
|
||||||
|
NetbriteConnectionException,
|
||||||
|
NetbriteTransferException,
|
||||||
|
Zone,
|
||||||
|
Message,
|
||||||
|
)
|
||||||
|
|
||||||
|
WIDTH = 300
|
||||||
|
DEFAULT_ZONE = Zone(0, 0, 150, 7)
|
||||||
|
DEFAULT_ZONE_NAME = "default"
|
||||||
|
|
||||||
|
|
||||||
|
@final
|
||||||
|
class SignControlGUI(tk.Tk):
|
||||||
|
def __init__(self):
|
||||||
|
super().__init__()
|
||||||
|
self.title("SignControl")
|
||||||
|
self.geometry(f"{WIDTH}x350")
|
||||||
|
self.resizable(False, True)
|
||||||
|
|
||||||
|
self.devices: list[NetBrite] = []
|
||||||
|
|
||||||
|
self._build_ui()
|
||||||
|
|
||||||
|
def _build_ui(self):
|
||||||
|
frm_devices = ttk.LabelFrame(self, text="Devices")
|
||||||
|
frm_devices.pack(fill="both", expand=True, padx=5, pady=5)
|
||||||
|
|
||||||
|
self.radios_frame = ttk.Frame(frm_devices)
|
||||||
|
self.radios_frame.pack(fill="both", expand=True, padx=5, pady=5)
|
||||||
|
|
||||||
|
self.selected_index = tk.IntVar(value=-1)
|
||||||
|
|
||||||
|
btn_frm = ttk.Frame(frm_devices)
|
||||||
|
btn_frm.pack(fill="both", padx=5, pady=5)
|
||||||
|
ttk.Button(btn_frm, text="Add", command=self._add_device).pack(side="left")
|
||||||
|
ttk.Button(btn_frm, text="Remove", command=self._remove_device).pack(
|
||||||
|
side="left"
|
||||||
|
)
|
||||||
|
|
||||||
|
frm_msg = ttk.LabelFrame(self, text="Message")
|
||||||
|
frm_msg.pack(fill="both", expand=False, padx=5, pady=5)
|
||||||
|
self.txt = tk.Text(frm_msg, height=1, wrap="word")
|
||||||
|
self.txt.pack(fill="both", expand=False, padx=5, pady=5)
|
||||||
|
|
||||||
|
hint = (
|
||||||
|
"Templates: {scrollon} {scrolloff} {blinkon} {blinkoff} "
|
||||||
|
"{left} {center} {right} {erase} {serialnum} {bell} "
|
||||||
|
"{red} {green} {yellow} {font <name>}"
|
||||||
|
"\nFonts: normal_7 normal_5 bold_7 monospace"
|
||||||
|
)
|
||||||
|
ttk.Label(
|
||||||
|
self,
|
||||||
|
text=hint,
|
||||||
|
font=("Segoe UI", 8),
|
||||||
|
foreground="gray",
|
||||||
|
wraplength=WIDTH - 20,
|
||||||
|
justify="center",
|
||||||
|
).pack(
|
||||||
|
padx=5,
|
||||||
|
pady=(0, 5),
|
||||||
|
)
|
||||||
|
|
||||||
|
btm_btn_frm = ttk.Frame(self)
|
||||||
|
btm_btn_frm.pack(fill="y", padx=5, pady=5)
|
||||||
|
ttk.Button(btm_btn_frm, text="Send", command=self._send_message).pack(
|
||||||
|
side="left"
|
||||||
|
)
|
||||||
|
ttk.Button(btm_btn_frm, text="Reset", command=self._reset_selected_zone).pack(
|
||||||
|
side="left"
|
||||||
|
)
|
||||||
|
ttk.Button(btm_btn_frm, text="Quit", command=self.quit).pack(side="right")
|
||||||
|
|
||||||
|
_ = self.bind("<Return>", lambda e: self._send_message())
|
||||||
|
|
||||||
|
def _get_selection(self) -> tuple[int, NetBrite] | None:
|
||||||
|
idx = self.selected_index.get()
|
||||||
|
if idx == -1:
|
||||||
|
_ = messagebox.showwarning("No device", "Please select a device first.")
|
||||||
|
return None
|
||||||
|
return idx, self.devices[idx]
|
||||||
|
|
||||||
|
def _add_device(self):
|
||||||
|
ip = simpledialog.askstring("Add device", "IP address:")
|
||||||
|
if not ip:
|
||||||
|
return
|
||||||
|
self._add_device_ip(ip.strip())
|
||||||
|
|
||||||
|
def _add_device_ip(self, ip: str):
|
||||||
|
try:
|
||||||
|
device = NetBrite(ip)
|
||||||
|
length = len(self.devices)
|
||||||
|
|
||||||
|
self._reset_zone(device)
|
||||||
|
self.devices.append(device)
|
||||||
|
|
||||||
|
radio = ttk.Radiobutton(
|
||||||
|
self.radios_frame,
|
||||||
|
text=ip,
|
||||||
|
value=length,
|
||||||
|
variable=self.selected_index,
|
||||||
|
)
|
||||||
|
radio.pack(anchor="w")
|
||||||
|
|
||||||
|
self.selected_index.set(length)
|
||||||
|
|
||||||
|
except NetbriteConnectionException as exc:
|
||||||
|
_ = messagebox.showerror("Connection error", str(exc))
|
||||||
|
|
||||||
|
def _remove_device(self):
|
||||||
|
sel = self._get_selection()
|
||||||
|
if sel is None:
|
||||||
|
return
|
||||||
|
idx, _ = sel
|
||||||
|
|
||||||
|
_ = self.devices.pop(idx)
|
||||||
|
|
||||||
|
radio = self.radios_frame.winfo_children()[idx]
|
||||||
|
radio.destroy()
|
||||||
|
|
||||||
|
for new_idx, widget in enumerate(self.radios_frame.winfo_children()):
|
||||||
|
if isinstance(widget, ttk.Radiobutton):
|
||||||
|
_ = widget.configure(value=new_idx)
|
||||||
|
|
||||||
|
self.selected_index.set(
|
||||||
|
-1 if not self.devices else min(idx, len(self.devices) - 1)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _send_message(self):
|
||||||
|
sel = self._get_selection()
|
||||||
|
if sel is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
_, dev = sel
|
||||||
|
raw_text = self.txt.get("1.0", "end-1c").strip()
|
||||||
|
if not raw_text:
|
||||||
|
return
|
||||||
|
|
||||||
|
msg = Message("{erase}" + raw_text)
|
||||||
|
try:
|
||||||
|
dev.message(msg, DEFAULT_ZONE_NAME)
|
||||||
|
self.txt.delete("1.0", "end")
|
||||||
|
except NetbriteTransferException as exc:
|
||||||
|
_ = messagebox.showerror(
|
||||||
|
"Send failed, try removing and adding the device.", str(exc)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _reset_selected_zone(self):
|
||||||
|
sel = self._get_selection()
|
||||||
|
if sel is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
_, dev = sel
|
||||||
|
self._reset_zone(dev)
|
||||||
|
|
||||||
|
def _reset_zone(self, device: NetBrite):
|
||||||
|
zone = DEFAULT_ZONE
|
||||||
|
message = Message("hello")
|
||||||
|
|
||||||
|
try:
|
||||||
|
device.zones({DEFAULT_ZONE_NAME: zone})
|
||||||
|
device.message(message, DEFAULT_ZONE_NAME)
|
||||||
|
except NetbriteTransferException as exc:
|
||||||
|
_ = messagebox.showerror(
|
||||||
|
"Send failed, try removing and adding the device.", str(exc)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
SignControlGUI().mainloop()
|
|
@ -15,7 +15,7 @@
|
||||||
nativeBuildInputs = [
|
nativeBuildInputs = [
|
||||||
pkgs.entr
|
pkgs.entr
|
||||||
pkgs.fastapi-cli
|
pkgs.fastapi-cli
|
||||||
(pkgs.python3.withPackages (x: [x.crc x.fastapi]))
|
(pkgs.python3.withPackages (x: [x.crc x.fastapi x.sqlmodel x.sqlalchemy x.tkinter]))
|
||||||
];
|
];
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|
28
netbrite.py
28
netbrite.py
|
@ -10,6 +10,14 @@ import re
|
||||||
DEFAULT_PORT = 700
|
DEFAULT_PORT = 700
|
||||||
|
|
||||||
|
|
||||||
|
class NetbriteConnectionException(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class NetbriteTransferException(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class Colors(Enum):
|
class Colors(Enum):
|
||||||
RED = 0x01
|
RED = 0x01
|
||||||
GREEN = 0x02
|
GREEN = 0x02
|
||||||
|
@ -101,7 +109,7 @@ class Message:
|
||||||
(rb"\{right\}", b"\x10\x28"),
|
(rb"\{right\}", b"\x10\x28"),
|
||||||
(rb"\{pause\}", b"\x10\x05"),
|
(rb"\{pause\}", b"\x10\x05"),
|
||||||
(rb"\{erase\}", b"\x10\x03"),
|
(rb"\{erase\}", b"\x10\x03"),
|
||||||
(rb"\{serial\}", b"\x10\x09"),
|
(rb"\{serialnum\}", b"\x10\x09"),
|
||||||
(rb"\{bell\}", b"\x10\x05"),
|
(rb"\{bell\}", b"\x10\x05"),
|
||||||
(rb"\{red\}", b"\x10\x0c" + pack("B", Colors.RED.value)),
|
(rb"\{red\}", b"\x10\x0c" + pack("B", Colors.RED.value)),
|
||||||
(rb"\{green\}", b"\x10\x0c" + pack("B", Colors.GREEN.value)),
|
(rb"\{green\}", b"\x10\x0c" + pack("B", Colors.GREEN.value)),
|
||||||
|
@ -183,16 +191,26 @@ class NetBrite:
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
self.sock.settimeout(5000)
|
self.sock.settimeout(2)
|
||||||
self.connect()
|
self.connect()
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
raise ConnectionError(f"Error while opening network socket. {e}")
|
raise NetbriteConnectionException(
|
||||||
|
f"Error while opening network socket. {e}"
|
||||||
|
)
|
||||||
|
|
||||||
def connect(self):
|
def connect(self):
|
||||||
self.sock.connect((self.address, self.port))
|
try:
|
||||||
|
self.sock.connect((self.address, self.port))
|
||||||
|
except OSError as e:
|
||||||
|
raise NetbriteConnectionException(
|
||||||
|
f"Error while opening network socket. {e}"
|
||||||
|
)
|
||||||
|
|
||||||
def tx(self, pkt: bytes):
|
def tx(self, pkt: bytes):
|
||||||
_ = self.sock.send(pkt)
|
try:
|
||||||
|
_ = self.sock.send(pkt)
|
||||||
|
except OSError as e:
|
||||||
|
raise NetbriteTransferException(f"Error while opening network socket. {e}")
|
||||||
|
|
||||||
def message(self, msg: Message, zoneName: str):
|
def message(self, msg: Message, zoneName: str):
|
||||||
z = self.zones_list.get(zoneName)
|
z = self.zones_list.get(zoneName)
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue