jsw/signcontrol
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Removed old perl from flake. cleaned up library just a little bit. added some message parsing

Browse source
This commit is contained in:
Jurn Wubben 2025-08-19 00:48:18 +02:00
parent 4e2aeffa8b
commit b3b19df77c
4 changed files with 84 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -1 +1,2 @@
.direnv
__pycache__/

6
flake.lock generated
View file

@ -2,11 +2,11 @@
"nodes": {
"nixpkgs": {
"locked": {
"lastModified": 1754381359,
"narHash": "sha256-fXmfK4eoA8hUqmEBWFlndH90Zc7vPjqDct+sjLXtCM8=",
"lastModified": 1755553648,
"narHash": "sha256-btiu4PUGLjoUgyHO5QoyxtMqd6WEkeLjgnI0XH16ea4=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "98b601302da9859b201bd858a8949aafec80ead1",
"rev": "f0ceb982688f1d75c0859ad43009bdb0eb22420d",
"type": "github"
},
"original": {

25
flake.nix
View file

@ -9,35 +9,12 @@
system = "x86_64-linux";
pkgs = nixpkgs.legacyPackages.${system};
in {
packages.${system}.default = let
pP = pkgs.perlPackages;
in
pkgs.perlPackages.buildPerlPackage rec {
pname = "SignControl";
version = "0.1";
src = ./.;
propagatedBuildInputs = [pP.DigestCRC pP.ModuleBuild];
nativeBuildInputs = [pkgs.shortenPerlShebang];
postInstall = ''
shortenPerlShebang $out/bin/signcontrol
'';
};
devShells.${system}.default =
pkgs.mkShell
{
nativeBuildInputs = [
(pkgs.perl.withPackages
(x: [
x.DigestCRC
x.ModuleBuild
]))
pkgs.entr
(pkgs.python3.withPackages (x: [x.crc]))
(pkgs.python3.withPackages (x: [x.crc x.fastapi]))
];
};
};

125
netbrite.py
View file

@ -55,6 +55,10 @@ def pkt_escape(pkt: bytes) -> bytes:
return bytes(buf)
COLORS = [i.name.lower() for i in Colors]
COLORS_PATTERH = rb"\{(" + "|".join(COLORS).encode("ascii") + rb")\}"
class Message:
activation_delay: int = 0 # Message activation delay
display_delay: int = 0 # Message display delay
@ -99,27 +103,27 @@ class Message:
(rb"\{erase\}", b"\x10\x03"),
(rb"\{serial\}", b"\x10\x09"),
(rb"\{bell\}", b"\x10\x05"),
(rb"\{red\}", b"\x10\x05" + pack("B", Colors.RED.value)),
(rb"\{green\}", b"\x10\x05" + pack("B", Colors.GREEN.value)),
(rb"\{yellow\}", b"\x10\x05" + pack("B", Colors.YELLOW.value)),
# (
# rb"\{(red|green|yellow)\}",
# lambda m: b"\x10\x0c" + pack("B", Colors[m[1].decode("ascii")].value),
# ),
# (
# rb"\{note\s+(\d+)\s+(\d+)\}",
# lambda m: b"\x10\x11" + pack("<HH", int(m[1]), int(m[2])),
# ),
# (
# rb"\{tune\s+([1-9])(\s+repeat)?\}",
# lambda m: bytes([0x10, 0x0A if m[2] else 0x0B, int(m[1])]),
# ),
# (
# rb"\{font\s+(\S+)\}",
# lambda m: bytes(
# [0x10, 0x0D, Fonts[m[1].decode("ascii").upper()].value]
# ),
# ),
(rb"\{red\}", b"\x10\x0c" + pack("B", Colors.RED.value)),
(rb"\{green\}", b"\x10\x0c" + pack("B", Colors.GREEN.value)),
(rb"\{yellow\}", b"\x10\x0c" + pack("B", Colors.YELLOW.value)),
(
COLORS_PATTERH,
lambda m: b"\x10\x0c" + pack("B", Colors[m[1].decode("ascii")].value),
),
(
rb"\{note\s+(\d+)\s+(\d+)\}",
lambda m: b"\x10\x11" + pack("<HH", int(m[1]), int(m[2])),
),
(
rb"\{tune\s+([1-9])(\s+repeat)?\}",
lambda m: bytes([0x10, 0x0A if m[2] else 0x0B, int(m[1])]),
),
(
rb"\{font\s+(\S+)\}",
lambda m: bytes(
[0x10, 0x0D, Fonts[m[1].decode("ascii").upper()].value]
),
),
]
for pat, repl in replacements:
@ -179,6 +183,7 @@ class NetBrite:
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(5000)
self.connect()
except OSError as e:
raise ConnectionError(f"Error while opening network socket. {e}")
@ -245,6 +250,7 @@ class NetBrite:
footer = pack("<HB", checksum(header + body), 0x04)
self.tx(pkt_escape(header + body + footer))
# print(f"Sent message to zone {zoneName}")
def zones(self, zones: dict[str, Zone] | None = None):
if zones != None:
@ -255,10 +261,9 @@ class NetBrite:
self.zones_list[zname].id = zid
z = self.zones_list[zname]
ztext = z.initial_text.parse_msg() # FIXME: parse_msg once implemented
ztext = z.initial_text.parse_msg()
zlen = len(ztext)
print(z.scroll_speed.value)
body = pack(
f"<4B B4B 3B BH 8B B 4B 4B H5B 10B 3B 20BH3B11B{zlen}s B",
0x0F, # Body start
@ -380,44 +385,54 @@ class NetBrite:
self.sessno += 1
_ = None
netbrt = NetBrite("10.65.37.244")
zones = [
# _ = None
# netbrt = NetBrite("10.65.37.244")
# zones = [
# Zone(
# 1,
# 0,
# 59,
# 50,
# 7,
# initial_text=Message("{erase}{scrolloff}{left}Welcome to:"),
# default_color=Colors.RED,
# ),
# Zone(
# 73,
# 1,
# 35,
# 7,
# initial_text=Message("{erase}{scrolloff}ING R&D"),
# initial_text=Message("{erase}{scrollon}{left}ING R&D"),
# default_color=Colors.YELLOW,
# default_font=Fonts.BOLD_7,
# scroll_speed=ScrollSpeeds.SLOW,
# ),
# # Zone(
# # 47,
# # 0,
# # 15,
# # 7,
# # initial_text=Message("{erase}{scrolloff}HQ"),
# # default_color=Colors.GREEN,
# # default_font=Fonts.BOLD_7,
# # ),
# Zone(
# 110,
# 80,
# 1,
# 10,
# 7,
# initial_text=Message("{erase}{scrolloff}HQ"),
# default_color=Colors.GREEN,
# 59,
# 6,
# initial_text=Message("{erase}{scrolloff}{left}Loading..."),
# default_color=Colors.RED,
# default_font=Fonts.NORMAL_5,
# ),
Zone(0, 0, 150, 7, initial_text=Message("{scrolloff}."))
]
netbrt.zones({str(k): v for k, v in enumerate(zones)})
from time import sleep, time
from datetime import datetime
while True:
now = datetime.now()
netbrt.message(
Message("{scrolloff}" + f"{now.hour}:{now.minute}:{now.second}"),
"0",
)
sleep(1)
# ]
# netbrt.zones({str(k): v for k, v in enumerate(zones)})
#
# from time import sleep, time
# from datetime import datetime
#
#
# def z(num: int) -> str:
# return str(num).rjust(2, "0")
#
#
# while True:
# now = datetime.now()
# netbrt.message(
# Message("{scrolloff}{left}" + f"{z(now.hour)}:{z(now.minute)}:{z(now.second)}"),
# "1",
# )
#
# t = time()
# sleep(1 - (t - int(t)))