added fonts hint

.gitignore

@@ -1,3 +1,2 @@
 .direnv
 __pycache__/
-devices.db

app.py

@@ -1,418 +1,176 @@
-from __future__ import annotations
-from contextlib import asynccontextmanager
-from time import sleep
-from typing import Annotated
-from fastapi import Depends, FastAPI, HTTPException
-from sqlmodel import SQLModel, Session, create_engine, select
+#!/usr/bin/env python3
 
-import netbrite as nb
-from db import (
-    MessageDB,
-    MessageUpdate,
-    NetBriteBase,
-    NetBriteDB,
-    NetBritePublic,
-    NetBriteUpdate,
-    ZoneBase,
-    ZoneDB,
-    ZonePublic,
-    ZoneUpdate,
+import tkinter as tk
+from tkinter import ttk, messagebox, simpledialog
+from typing import final
+from netbrite import (
+    NetBrite,
+    NetbriteConnectionException,
+    NetbriteTransferException,
+    Zone,
+    Message,
 )
 
-DB_URL = "sqlite:///devices.db"
-engine = create_engine(DB_URL, connect_args={"check_same_thread": False})
+WIDTH = 300
+DEFAULT_ZONE = Zone(0, 0, 150, 7)
+DEFAULT_ZONE_NAME = "default"
 
 
-def get_session():
-    with Session(engine) as session:
-        yield session
+@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] = []
 
-SessionDep = Annotated[Session, Depends(get_session)]
+        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)
 
-@asynccontextmanager
-async def lifespan(_: FastAPI):
-    SQLModel.metadata.create_all(engine)
-    load_devices_from_db()
-    yield
+        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)
 
-app = FastAPI(lifespan=lifespan)
-active_devices: dict[int, nb.NetBrite] = {}
+        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)
 
-# ---------- helper ----------
-def load_devices_from_db() -> None:
-    with Session(engine) as session:
-        for device in session.exec(select(NetBriteDB)).all():
-            load_device(device)
+        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")
 
-def load_device(device: NetBriteDB):
-    id = device.id or 0
-    try:
-        active_devices[id] = nb.NetBrite(device.address, device.port)
+        _ = self.bind("<Return>", lambda e: self._send_message())
 
-        load_zones(device.zones, active_devices[id])
-    except nb.NetbriteConnectionException as exc:
-        print(f"Could not connect to {device.address}:{device.port} — {exc}")
+    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 load_zones_id(session: Session, device_id: int, net_dev: nb.NetBrite):
-    statement = select(ZoneDB).where(ZoneDB.netbrite_id == device_id)
-    zones = list(session.exec(statement))
-    load_zones(zones, net_dev)
+    def _add_device_ip(self, ip: str):
+        try:
+            device = NetBrite(ip)
+            length = len(self.devices)
 
+            self._reset_zone(device)
+            self.devices.append(device)
 
-def load_zones(zones_in: list[ZoneDB], net_dev: nb.NetBrite) -> None:
-    zones: dict[str, nb.Zone] = {}
-    messages: dict[str, nb.Message] = {}
-
-    for zone in zones_in:
-        msg = zone.default_message
-
-        default_msg = (
-            nb.Message(
-                activation_delay=msg.activation_delay,
-                display_delay=msg.display_delay,
-                priority=msg.priority,
-                text=msg.text,
-                ttl=msg.ttl,
+            radio = ttk.Radiobutton(
+                self.radios_frame,
+                text=ip,
+                value=length,
+                variable=self.selected_index,
             )
-            if msg
-            else nb.Message(f"Zone {zone.name}")
+            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)
         )
 
-        zones[zone.name] = nb.Zone(
-            x=zone.x,
-            y=zone.y,
-            width=zone.width,
-            height=zone.height,
-            scroll_speed=zone.scroll_speed,
-            pause_duration=zone.pause_duration,
-            volume=zone.volume,
-            default_font=zone.default_font,
-            default_color=zone.default_color,
-            initial_text=default_msg,
-        )
-        messages[zone.name] = default_msg
-        # print(f"{zone.name}: {default_msg.text}")
+    def _send_message(self):
+        sel = self._get_selection()
+        if sel is None:
+            return
 
-    if zones:
-        net_dev.zones(zones)
-        sleep(0.2)
+        _, dev = sel
+        raw_text = self.txt.get("1.0", "end-1c").strip()
+        if not raw_text:
+            return
 
-    for zone, message in messages.items():
-        net_dev.message(message, zone)
-        sleep(0.2)
-
-
-def create_default_zone(session: Session, device_id: int) -> None:
-    zone = ZoneDB(
-        name="0",
-        netbrite_id=device_id,
-    )
-    msg = MessageDB(
-        text="{erase}Welcome",
-    )
-    session.add(msg)
-    session.add(zone)
-
-    session.commit()
-    session.refresh(zone)
-    session.refresh(msg)
-
-    zone.default_message_id = msg.id
-    session.commit()
-
-
-# ---------- routes ----------
-@app.post("/api/devices", response_model=NetBritePublic)
-def create_device(device: NetBriteBase, session: SessionDep):
-    if session.exec(
-        select(NetBriteDB).where(NetBriteDB.address == device.address)
-    ).first():
-        raise HTTPException(400, "Device already exists")
-
-    db_device = NetBriteDB.model_validate(device)
-    session.add(db_device)
-    session.commit()
-    session.refresh(db_device)
-
-    create_default_zone(session, db_device.id or 0)
-
-    load_device(db_device)
-    return db_device
-
-
-@app.get("/api/devices", response_model=list[NetBritePublic])
-def get_devices(session: SessionDep):
-    devices: list[NetBritePublic] = []
-
-    for device in session.exec(select(NetBriteDB)).all():
-        device = NetBritePublic.model_validate(
-            device, update={"active": (device.id or 0) in active_devices}
-        )
-        devices.append(device)
-
-    return devices
-
-
-@app.post(
-    "/api/devices/{device_id}",
-    response_model=NetBritePublic,
-    description="**NOTE**: this **WILL** disconnect the device. You'll have to manually reconnect it.",
-)
-def edit_device(device_id: int, updated_device: NetBriteUpdate, session: SessionDep):
-    db_dev = session.get(NetBriteDB, device_id)
-    if not db_dev:
-        raise HTTPException(404, "Device not found")
-
-    if device_id in active_devices:
+        msg = Message("{erase}" + raw_text)
         try:
-            active_devices[device_id].sock.close()
-        except OSError:
-            print("Failed to close socket.")
+            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)
+            )
 
-        del active_devices[device_id]
+    def _reset_selected_zone(self):
+        sel = self._get_selection()
+        if sel is None:
+            return
 
-    dev_data = updated_device.model_dump(exclude_unset=True)
-    _ = db_dev.sqlmodel_update(dev_data)
+        _, dev = sel
+        self._reset_zone(dev)
 
-    session.add(db_dev)
-    session.commit()
-    session.refresh(db_dev)
+    def _reset_zone(self, device: NetBrite):
+        zone = DEFAULT_ZONE
+        message = Message("hello")
 
-    return db_dev
-
-
-@app.delete("/api/devices/{device_id}")
-def delete_device(device_id: int, session: SessionDep):
-    db_dev = session.get(NetBriteDB, device_id)
-    if not db_dev:
-        raise HTTPException(404, "Device not found")
-
-    delete: list[MessageDB | ZoneDB | NetBriteDB] = [db_dev]
-    for zone in db_dev.zones:
-        if zone.default_message != None:
-            delete.append(zone.default_message)
-        delete.append(zone)
-
-    if device_id in active_devices:
         try:
-            active_devices[device_id].sock.close()
-        except OSError:
-            print("Failed to close socket.")
-        del active_devices[device_id]
-
-    for i in delete:
-        session.delete(i)
-
-    session.commit()
-    return 200
+            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)
+            )
 
 
-@app.post("/api/devices/{device_id}/reconnect")
-def reconnect_device(device_id: int, session: SessionDep):
-    db_dev = session.get(NetBriteDB, device_id)
-    if not db_dev:
-        raise HTTPException(404, "Device not found")
-
-    try:
-        new_netbrite = nb.NetBrite(db_dev.address, db_dev.port)
-        active_devices[device_id] = new_netbrite
-        load_zones_id(session, device_id, active_devices[device_id])
-        return 200
-    except nb.NetbriteConnectionException as exc:
-        raise HTTPException(400, str(exc))
-
-
-@app.post(
-    "/api/devices/{device_id}/sync",
-    description="**NOTE**: This will recreate the zones on the device.",
-)
-def sync_device(
-    device_id: int,
-    session: SessionDep,
-):
-    if device_id not in active_devices:
-        raise HTTPException(500, "Device not active, try reconnecting")
-
-    try:
-        load_zones_id(session, device_id, active_devices[device_id])
-    except nb.NetbriteTransferException:
-        del active_devices[device_id]
-        raise HTTPException(500, "Failed to send zones. Device inactive now.")
-
-    return 200
-
-
-@app.post(
-    "/api/devices/{device_id}/restart",
-)
-def restart_device(
-    device_id: int,
-):
-    if device_id not in active_devices:
-        raise HTTPException(500, "Device not active, try reconnecting")
-
-    try:
-        active_devices[device_id].reboot()
-        del active_devices[device_id]
-    except nb.NetbriteTransferException:
-        raise HTTPException(500, "Failed to send reboot command. Device inactive now.")
-
-    return 200
-
-
-@app.post(
-    "/api/devices/{device_id}/zones",
-    response_model=ZonePublic,
-    description="**NOTE**: this does not update the device.",
-)
-def create_zone(device_id: int, new_zone: ZoneBase, session: SessionDep):
-    device = session.get(NetBriteDB, device_id)
-    if not device:
-        raise HTTPException(404, "Device not found")
-
-    new_zone_data = new_zone.model_dump(exclude_unset=True)
-    extra_data = {"netbrite_id": device_id}
-    zone = ZoneDB.model_validate(new_zone_data, update=extra_data)
-
-    msg = MessageDB(
-        text="{erase}Welcome",
-    )
-    session.add(zone)
-    session.add(msg)
-
-    session.commit()
-    session.refresh(zone)
-    session.refresh(msg)
-
-    zone.default_message_id = msg.id
-    session.commit()
-
-    return zone
-
-
-@app.get("/api/devices/{device_id}/zones", response_model=list[ZonePublic])
-def get_zones(device_id: int, session: SessionDep):
-    device = session.get(NetBriteDB, device_id)
-    if not device:
-        raise HTTPException(404, "Device not found")
-
-    return device.zones
-
-
-@app.delete(
-    "/api/zone/{zone_id}",
-    description="**NOTE**: this does not update the device.",
-)
-def delete_zone(
-    zone_id: int,
-    session: SessionDep,
-):
-    zone = session.get(ZoneDB, zone_id)
-    if not zone:
-        raise HTTPException(404, "Zone not found")
-
-    message = zone.default_message
-    if message:
-        session.delete(message)
-
-    session.delete(zone)
-    session.commit()
-
-    return 200
-
-
-@app.post(
-    "/api/zone/{zone_id}",
-    response_model=ZonePublic,
-    description="**NOTE**: this does not update the device.",
-)
-def edit_zone(zone_id: int, zone: ZoneUpdate, session: SessionDep):
-    zone_db = session.get(ZoneDB, zone_id)
-
-    if not zone_db:
-        raise HTTPException(404, "Zone not found")
-
-    zone_update_data = zone.model_dump(exclude_unset=True)
-    print(zone_update_data)
-    _ = zone_db.sqlmodel_update(zone_update_data)
-
-    session.add(zone_db)
-    session.commit()
-    session.refresh(zone_db)
-
-    return zone_db
-
-
-@app.post(
-    "/api/zone/{zone_id}/message",
-    response_model=ZonePublic,
-    description="**NOTE**: this does not update the device.",
-)
-def edit_message(zone_id: int, message: MessageUpdate, session: SessionDep):
-    zone_db = session.get(ZoneDB, zone_id)
-
-    if not zone_db:
-        raise HTTPException(404, "Zone not found")
-
-    if not zone_db.default_message:
-        db_message = MessageDB.model_validate(message)
-
-        session.add(db_message)
-        session.commit()
-        session.refresh(db_message)
-
-        zone_db.default_message_id = db_message.id
-        session.add(zone_db)
-        session.commit()
-        session.refresh(zone_db)
-    else:
-        db_message = zone_db.default_message
-        data = message.model_dump(
-            exclude_unset=True,
-        )
-        _ = db_message.sqlmodel_update(data, update={"id": db_message.id})
-
-        session.add(db_message)
-        session.commit()
-        session.refresh(zone_db)
-
-    return zone_db
-
-
-@app.post(
-    "/api/zone/{zone_id}/adhoc_message",
-    description="**NOTE**: this updates the device temporarily. Edit the zone message for permanent changes.",
-)
-def adhoc_message(zone_id: int, message: MessageUpdate, session: SessionDep):
-    zone_db = session.get(ZoneDB, zone_id)
-
-    if not zone_db:
-        raise HTTPException(404, "Zone not found")
-
-    device_id = zone_db.netbrite_id
-    if not device_id in active_devices:
-        raise HTTPException(500, "Device inactive")
-
-    nb_message = nb.Message(
-        text=message.text or "",
-        activation_delay=message.activation_delay or 0,
-        display_delay=message.display_delay or 0,
-        display_repeat=message.display_repeat or 0,
-        priority=message.priority or nb.Priorities.OVERRIDE,
-        sound_alarm=message.sound_alarm or False,
-        ttl=message.ttl or 0,
-    )
-
-    try:
-        active_devices[device_id].message(nb_message, zone_db.name)
-    except nb.NetbriteTransferException:
-        del active_devices[device_id]
-        raise HTTPException(500, "Failed to send message. Device inactive now.")
-
-    return 200
+if __name__ == "__main__":
+    SignControlGUI().mainloop()

db.py

@@ -1,111 +0,0 @@
-# from __future__ import annotations
-from sqlmodel import Field, Relationship, SQLModel
-from netbrite import Colors, Fonts, Priorities, ScrollSpeeds, Message
-
-MAX_WIDTH = 120
-MAX_HEIGHT = 7
-
-
-# --- Message ---
-class MessageBase(SQLModel):
-    text: str = ""
-    activation_delay: int = 0
-    display_delay: int = 0
-    display_repeat: int = 0
-    priority: Priorities = Priorities.OVERRIDE
-    sound_alarm: bool = False
-    ttl: int = 0
-
-
-class MessageUpdate(SQLModel):
-    text: str | None = ""
-    activation_delay: int | None = 0
-    display_delay: int | None = 0
-    display_repeat: int | None = 0
-    priority: Priorities | None = Priorities.OVERRIDE
-    sound_alarm: bool = False
-    ttl: int | None = 0
-
-
-class MessageDB(MessageBase, table=True):
-    id: int | None = Field(default=None, primary_key=True)
-    zone: "ZoneDB" = Relationship(  # pyright: ignore[reportAny]
-        back_populates="default_message"
-    )
-
-
-class MessagePublic(MessageBase):
-    id: int
-
-
-# --- Device ---
-class NetBriteBase(SQLModel):
-    address: str = Field(unique=True, index=True)
-    port: int = 700
-
-
-class NetBriteUpdate(SQLModel):
-    address: str = Field(unique=True, index=True)
-    port: int = 700
-
-
-class NetBriteDB(NetBriteBase, table=True):
-    id: int | None = Field(default=None, primary_key=True)
-    zones: list["ZoneDB"] = Relationship(  # pyright: ignore[reportAny]
-        back_populates="netbrite"
-    )
-
-
-class NetBritePublic(NetBriteBase):
-    id: int
-    zones: list["ZoneDB"]
-    active: bool
-
-
-# --- Zone ---
-class ZoneBase(SQLModel):
-    name: str
-    x: int = 0
-    y: int = 0
-    width: int = MAX_WIDTH
-    height: int = MAX_HEIGHT
-    scroll_speed: ScrollSpeeds = ScrollSpeeds.NORMAL
-    pause_duration: int = 1000
-    volume: int = 4
-    default_font: Fonts = Fonts.NORMAL_7
-    default_color: Colors = Colors.RED
-
-
-class ZoneUpdate(SQLModel):
-    name: str | None = None
-    x: int | None = None
-    y: int | None = None
-    width: int | None = None
-    height: int | None = None
-    scroll_speed: ScrollSpeeds | None = ScrollSpeeds.NORMAL
-    pause_duration: int | None = 1000
-    volume: int | None = 4
-    default_font: Fonts | None = Fonts.NORMAL_7
-    default_color: Colors | None = Colors.RED
-
-
-class ZoneDBBase(ZoneBase):
-    default_message_id: int | None = Field(default=None, foreign_key="messagedb.id")
-    netbrite_id: int = Field(default=None, foreign_key="netbritedb.id")
-
-
-class ZoneDB(ZoneDBBase, table=True):
-    id: int | None = Field(default=None, primary_key=True)
-
-    default_message: MessageDB | None = Relationship(  # pyright: ignore[reportAny]
-        back_populates="zone"
-    )
-    netbrite: NetBriteDB = Relationship(  # pyright: ignore[reportAny]
-        back_populates="zones"
-    )
-
-
-class ZonePublic(ZoneDBBase):
-    id: int
-    default_message: MessagePublic
-    # netbrite: NetBritePublic

flake.nix

@@ -15,7 +15,7 @@
         nativeBuildInputs = [
           pkgs.entr
           pkgs.fastapi-cli
-          (pkgs.python3.withPackages (x: [x.crc x.fastapi x.sqlmodel x.sqlalchemy]))
+          (pkgs.python3.withPackages (x: [x.crc x.fastapi x.sqlmodel x.sqlalchemy x.tkinter]))
         ];
       };
   };

netbrite.py

@@ -1,4 +1,3 @@
-import select
 from typing import Callable
 from crc import Calculator, Crc16
 from enum import Enum
@@ -64,13 +63,6 @@ def pkt_escape(pkt: bytes) -> bytes:
     return bytes(buf)
 
 
-def is_socket_alive(sock: SocketType):
-    readable, _, exceptional = select.select([sock], [], [sock], 1)
-    if sock in exceptional:
-        return False
-    return bool(readable)
-
-
 COLORS = [i.name.lower() for i in Colors]
 COLORS_PATTERH = rb"\{(" + "|".join(COLORS).encode("ascii") + rb")\}"
 
@@ -208,8 +200,6 @@ class NetBrite:
 
     def connect(self):
         try:
-            if not is_socket_alive(self.sock):
-                raise OSError("Socket dead")
             self.sock.connect((self.address, self.port))
         except OSError as e:
             raise NetbriteConnectionException(
@@ -218,37 +208,10 @@ class NetBrite:
 
     def tx(self, pkt: bytes):
         try:
-            _ = self.sock.sendall(pkt_escape(pkt))
+            _ = self.sock.send(pkt)
         except OSError as e:
             raise NetbriteTransferException(f"Error while opening network socket. {e}")
 
-    def reboot(self):
-        pkt = pack(
-            f"<3B H H 3B 2B 4B 4B 1B",
-            0x16,
-            0x16,
-            0x01,  # msg start
-            2,  # body length
-            self.seqno,  # packet count
-            0x00,
-            0x01,
-            0x00,
-            0x01,
-            0x01,  # msg type: reset
-            0x00,
-            0xC8,
-            0x01,
-            0x00,  # sign id
-            0x00,
-            0x01,
-            0x0F,
-            0x00,  # reset msg
-            0x17,  # crc follows
-        )
-        pkt += pack("<HB", checksum(pkt), 0x04)
-
-        self.tx(pkt)
-
     def message(self, msg: Message, zoneName: str):
         z = self.zones_list.get(zoneName)
         if z == None:
@@ -304,7 +267,7 @@ class NetBrite:
         )
 
         footer = pack("<HB", checksum(header + body), 0x04)
-        self.tx(header + body + footer)
+        self.tx(pkt_escape(header + body + footer))
         # print(f"Sent message to zone {zoneName}")
 
     def zones(self, zones: dict[str, Zone] | None = None):
@@ -319,22 +282,6 @@ class NetBrite:
             ztext = z.initial_text.parse_msg()
             zlen = len(ztext)
 
-            rect: list[int] = list(z.rect)
-
-            if rect[0] > 254:
-                rect[0] = 254
-
-            if rect[1] > 254:
-                rect[1] = 254
-
-            if rect[0] >= rect[2]:
-                rect[2] = rect[0] + 1
-
-            if rect[1] >= rect[3]:
-                rect[3] = rect[1] + 1
-
-            z.rect = tuple(rect)  # pyright: ignore[reportAttributeAccessIssue]
-
             body = pack(
                 f"<4B B4B 3B BH 8B B 4B 4B H5B 10B 3B 20BH3B11B{zlen}s B",
                 0x0F,  # Body start
@@ -449,7 +396,7 @@ class NetBrite:
 
             footer = pack("<HB", crc, 0x04)
 
-            self.tx(header + body + footer)
+            self.tx(pkt_escape(header + body + footer))
             print(f"Sent zone {zname}")
 
             self.seqno += 1