212 lines
6 KiB
Python
212 lines
6 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import math
|
|
import time
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import serial
|
|
import click
|
|
|
|
seg_map = {
|
|
" ": 0x00,
|
|
"a": 0x2e,
|
|
"b": 0xd6,
|
|
"c": 0xd0,
|
|
"d": 0xc5,
|
|
"e": 0x59,
|
|
"f": 0x98,
|
|
"g": 0xd4,
|
|
"h": 0x8c,
|
|
"i": 0x5c,
|
|
"j": 0x78,
|
|
"k": 0x8e,
|
|
"l": 0xc0,
|
|
"m": 0xa3,
|
|
"n": 0xa5,
|
|
"o": 0xf0,
|
|
"p": 0x93,
|
|
"q": 0xf4,
|
|
"r": 0x9e,
|
|
"s": 0x55,
|
|
"t": 0xc8,
|
|
"u": 0xe0,
|
|
"v": 0x8a,
|
|
"w": 0xac,
|
|
"x": 0x0f,
|
|
"y": 0x0b,
|
|
"z": 0x5a,
|
|
"0": 0x13,
|
|
"1": 0x20,
|
|
"2": 0x16,
|
|
"3": 0x56,
|
|
"4": 0x23,
|
|
"5": 0x1c,
|
|
"6": 0x4e,
|
|
"7": 0x1a,
|
|
"8": 0x5f,
|
|
"9": 0x33,
|
|
"/": 0x0a,
|
|
"\\": 0x05,
|
|
".": 0x04,
|
|
",": 0x08,
|
|
"_": 0x40,
|
|
"!": 0x53,
|
|
"?": 0x52,
|
|
"*": 0x0f,
|
|
":": 0x50,
|
|
"(": 0x06,
|
|
"<": 0x06,
|
|
"[": 0xd0,
|
|
")": 0x09,
|
|
">": 0x09,
|
|
"]": 0x70,
|
|
"|": 0x20,
|
|
"#": 0xff,
|
|
}
|
|
|
|
def shuffle_segments(d):
|
|
bits_in = [7, 6, 5, 4, 3, 2, 1, 0]
|
|
bits_out = [7, 6, 5, 4, 3, 2, 1, 0]
|
|
|
|
out = 0
|
|
for bit_in, bit_out in zip(bits_in, bits_out):
|
|
if d & (1<<bit_in):
|
|
out |= 1<<bit_out
|
|
return out
|
|
|
|
digit_addresses = [
|
|
(0, 0), # 0
|
|
(0, 1), # 1
|
|
(0, 2), # 2
|
|
(0, 3), # 3
|
|
(0, 4), # 4
|
|
(0, 5), # 5
|
|
(0, 6), # 6
|
|
(0, 7), # 7
|
|
(1, 8), # 8
|
|
(1, 9), # 9
|
|
(1, 10), # 10
|
|
(1, 11), # 11
|
|
(1, 12), # 12
|
|
(1, 13), # 13
|
|
(1, 14), # 14
|
|
(1, 15), # 15
|
|
(2, 0), # 16
|
|
(2, 1), # 17
|
|
(2, 2), # 18
|
|
(2, 3), # 19
|
|
(2, 4), # 20
|
|
(2, 5), # 21
|
|
(2, 6), # 22
|
|
(2, 7), # 23
|
|
(3, 8), # 24
|
|
(3, 9), # 25
|
|
(3, 10), # 26
|
|
(3, 11), # 27
|
|
(3, 12), # 28
|
|
(3, 13), # 29
|
|
(3, 14), # 30
|
|
(3, 15), # 31
|
|
]
|
|
|
|
def log(*args):
|
|
timestamp = datetime.now().strftime('[%y-%m-%d %H:%M:%S]')
|
|
print(timestamp, *args)
|
|
|
|
def set_display(ser, message, interval, display_width, global_brightness):
|
|
lines = [ line for line in message.splitlines() if line.strip() ]
|
|
for i, line in enumerate(lines):
|
|
line = line.center(display_width)[:display_width]
|
|
line_mapped = [shuffle_segments(seg_map.get(c, seg_map['#'])) for c in line]
|
|
log(f' [Line {i+1}/{len(lines)}] Sending control data.')
|
|
|
|
digit_data = [0x00] * 16 * 4
|
|
for (driver, address), data in zip(digit_addresses, line_mapped):
|
|
digit_data[driver*16 + address] = data
|
|
|
|
formatted = ''.join(f'{global_brightness:1x}{d:02x}' for d in digit_data) + '\n'
|
|
if ser is not None:
|
|
ser.write(formatted)
|
|
ser.flush()
|
|
time.sleep(interval)
|
|
|
|
@click.command()
|
|
@click.option('--dry-run/--wet-run')
|
|
@click.option('-b', '--baudrate', type=int, default=9600)
|
|
@click.option('-w', '--display-width', type=int, default=32)
|
|
@click.option('-m', '--message-interval', type=float, default=30)
|
|
@click.option('-g', '--global-brightness', type=click.IntRange(0, 15), default=15)
|
|
@click.option('-r', '--idle-rotation', type=click.Path(dir_okay=False, path_type=Path), default='idle_rotation.txt')
|
|
@click.argument('database')
|
|
@click.argument('port', default=None, required=False)
|
|
def cli(database, port, baudrate, message_interval, display_width, global_brightness, idle_rotation, dry_run):
|
|
db = sqlite3.connect(database)
|
|
ser = serial.Serial(port, baudrate) if port is not None and not dry_run else None
|
|
|
|
idle_index = 0
|
|
idle_seconds = 0
|
|
discard = 0
|
|
while True:
|
|
count, = db.execute('''
|
|
SELECT COUNT(*) FROM messages
|
|
WHERE suppress_display = 0 AND timestamp_displayed IS NULL
|
|
ORDER BY timestamp_received''').fetchone()
|
|
|
|
if count == 0:
|
|
if idle_seconds > 0:
|
|
idle_seconds -= 1
|
|
else:
|
|
if idle_rotation.is_file():
|
|
idle_messages = idle_rotation.read_text().splitlines()
|
|
else:
|
|
idle_messages = ['10:37C3', '10:UNLOCKED']
|
|
|
|
if idle_index >= len(idle_messages):
|
|
idle_index = 0
|
|
|
|
interval, _, message = idle_messages[idle_index].partition(':')
|
|
idle_seconds = int(interval)
|
|
idle_index += 1
|
|
|
|
log(f' [Idle {idle_index}/{len(idle_messages)}] {message}')
|
|
set_display(ser, message, 0, display_width, global_brightness)
|
|
|
|
time.sleep(1)
|
|
continue
|
|
|
|
queue_pressure = max(0, min((math.log10(max(1, count-3)) - 0.3) * message_interval*2/3, message_interval*1/3))
|
|
interval = message_interval - queue_pressure
|
|
log(f' Waiting interval of {message_interval:.1f} s with queue pressure {queue_pressure:.1f} from n={count} items for a total of {interval:.1f} s')
|
|
|
|
|
|
rowid, message, timestamp, remote_ip, suppress_display = db.execute('''
|
|
SELECT rowid, message, timestamp_received, remote_ip, suppress_display FROM messages
|
|
WHERE timestamp_displayed IS NULL
|
|
ORDER BY timestamp_received LIMIT 1
|
|
''').fetchone()
|
|
timestamp = datetime.fromisoformat(timestamp)
|
|
delta = (datetime.utcnow() - timestamp).total_seconds()
|
|
|
|
db.execute('UPDATE messages SET timestamp_displayed = datetime("now") WHERE rowid=?', (rowid,))
|
|
db.commit()
|
|
|
|
if suppress_display:
|
|
discard += 1
|
|
|
|
else:
|
|
if discard > 0:
|
|
log(f'Discarded {discard} messages.')
|
|
discard = 0
|
|
|
|
log(f'Queueing message by {remote_ip} received at {timestamp} ({delta//60:.0f}m {delta%60:.0f}s ago):')
|
|
log(' ', repr(message))
|
|
|
|
idle_seconds = 0
|
|
interval = max(5, interval/len(message.splitlines()))
|
|
set_display(ser, message, interval, display_width, global_brightness)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
cli()
|