diff --git a/app.py b/app.py new file mode 100644 index 0000000..76efb1f --- /dev/null +++ b/app.py @@ -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 }" + "\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("", 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() diff --git a/flake.nix b/flake.nix index 53bf9e0..5739b33 100644 --- a/flake.nix +++ b/flake.nix @@ -15,7 +15,7 @@ nativeBuildInputs = [ pkgs.entr 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])) ]; }; }; diff --git a/netbrite.py b/netbrite.py index ba97efb..1a39dbb 100644 --- a/netbrite.py +++ b/netbrite.py @@ -10,6 +10,14 @@ import re DEFAULT_PORT = 700 +class NetbriteConnectionException(Exception): + pass + + +class NetbriteTransferException(Exception): + pass + + class Colors(Enum): RED = 0x01 GREEN = 0x02 @@ -101,7 +109,7 @@ class Message: (rb"\{right\}", b"\x10\x28"), (rb"\{pause\}", b"\x10\x05"), (rb"\{erase\}", b"\x10\x03"), - (rb"\{serial\}", b"\x10\x09"), + (rb"\{serialnum\}", b"\x10\x09"), (rb"\{bell\}", b"\x10\x05"), (rb"\{red\}", b"\x10\x0c" + pack("B", Colors.RED.value)), (rb"\{green\}", b"\x10\x0c" + pack("B", Colors.GREEN.value)), @@ -183,16 +191,26 @@ class NetBrite: try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.sock.settimeout(5000) + self.sock.settimeout(2) self.connect() 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): - 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): - _ = 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): z = self.zones_list.get(zoneName)