LingSys

#!/usr/bin/env python3
"""Show a window (hatchmini by default) on a Thermalright-style USB LCD, in real time.

    python tools/lcd_sender.py                    mirror the "LingSys Mini" window to the display
    python tools/lcd_sender.py --test             send a test card and exit (check the cable, size and rotation)
    python tools/lcd_sender.py --rotate 270       turn the picture (0, 90, 180, 270)
    python tools/lcd_sender.py --title "Some Window" --fps 8 --quality 80

The display (USB 0416:5302, "USBDISPLAY") takes JPEG frames. This is the protocol TRCC's own helper
(USBLCDNEW) uses, read out of TRCC's code: a handshake packet, then for each frame a 20-byte header

    DA DB DC DD | 02 00 00 00 (picture) | width u16 | height u16 | 02 00 00 00 (JPEG) | JPEG length u32

followed by the JPEG, all sent in 512-byte chunks to endpoint 2 of the HID interface. The panel answers the
handshake with a code (yours: 58, a 240 x 320 portrait panel; TRCC turns its 320 x 240 pictures 90 degrees
to fit it). Only one program can own the display, so close TRCC first (run.ps1 lcd does that for you).

Needs: pip install pyusb libusb-package pillow
"""
import argparse, ctypes, io, sys, time
from ctypes import wintypes

import usb.core, usb.util
import usb.backend.libusb1 as libusb1
import libusb_package
from PIL import Image, ImageDraw, ImageGrab

VID, PID = 0x0416, 0x5302
EP_OUT, EP_IN = 0x02, 0x81

# panel code from the handshake -> (width, height) of the frames the panel wants.
# Only the codes whose packet format I read out of TRCC are listed; 58 is the one tested here.
PANELS = {58: (240, 320), 72: (480, 480), 129: (480, 480), 54: (360, 360), 128: (1280, 480)}
DEFAULT_PANEL = (240, 320)


def open_device():
    backend = libusb1.get_backend(find_library=libusb_package.find_library)
    dev = usb.core.find(idVendor=VID, idProduct=PID, backend=backend)
    if dev is None:
        sys.exit('display not found (USB %04x:%04x). Is it plugged in, and is TRCC closed?' % (VID, PID))
    try:
        dev.set_configuration(1)
    except usb.core.USBError:
        pass
    try:
        usb.util.claim_interface(dev, 0)
    except usb.core.USBError as e:
        sys.exit('could not claim the display (%s). Close TRCC and its USBLCD helpers first.' % e)
    return dev


def handshake(dev):
    pkt = bytes([0xDA, 0xDB, 0xDC, 0xDD, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]) + bytes(492)
    dev.write(EP_OUT, pkt, timeout=1000)
    r = bytes(dev.read(EP_IN, 512, timeout=1500))
    if r[:4] != bytes([0xDA, 0xDB, 0xDC, 0xDD]) or len(r) < 21 or r[12] != 1:
        raise RuntimeError('unexpected handshake reply: ' + r[:40].hex(' '))
    serial = r[20:27].decode('ascii', 'replace') if len(r) >= 27 else ''
    return r[5], r[4], serial          # panel code, sub code, serial


def make_packet(jpeg, size):
    w, h = size
    hdr = bytearray([0xDA, 0xDB, 0xDC, 0xDD, 2, 0, 0, 0]) + w.to_bytes(2, 'little') + h.to_bytes(2, 'little') \
        + bytes([2, 0, 0, 0]) + len(jpeg).to_bytes(4, 'little')
    data = bytes(hdr) + jpeg
    pad = (-len(data)) % 512
    return data + bytes(pad)


def send_frame(dev, packet):
    for i in range(0, len(packet), 512):
        dev.write(EP_OUT, packet[i:i + 512], timeout=1000)


def encode(img, size, rotate, quality):
    """img: RGB picture in the screen's own orientation (e.g. 320 x 240). Returns JPEG bytes sized for the panel."""
    if rotate:
        img = img.rotate(-rotate, expand=True)      # clockwise, like TRCC's RotateImg
    if img.size != size:
        img = img.resize(size, Image.LANCZOS)
    buf = io.BytesIO()
    img.save(buf, 'JPEG', quality=quality, optimize=False)
    data = buf.getvalue()
    while len(data) >= 450000 and quality > 20:     # TRCC lowers the quality the same way
        quality -= 5
        buf = io.BytesIO()
        img.save(buf, 'JPEG', quality=quality)
        data = buf.getvalue()
    return data


# ---- grabbing a window by title (Windows)
user32 = ctypes.windll.user32
gdi32 = ctypes.windll.gdi32


def find_window(title):
    return user32.FindWindowW(None, title)


def client_size(hwnd):
    r = wintypes.RECT()
    user32.GetClientRect(hwnd, ctypes.byref(r))
    return r.right - r.left, r.bottom - r.top


def grab_window(hwnd):
    """The window's client area as an RGB image, even if other windows cover it (falls back to a screen grab)."""
    w, h = client_size(hwnd)
    if w <= 0 or h <= 0:
        return None
    hdc = user32.GetDC(hwnd)
    mem = gdi32.CreateCompatibleDC(hdc)
    bmp = gdi32.CreateCompatibleBitmap(hdc, w, h)
    old = gdi32.SelectObject(mem, bmp)
    ok = user32.PrintWindow(hwnd, mem, 3)           # PW_CLIENTONLY | PW_RENDERFULLCONTENT
    img = None
    if ok:
        class BMI(ctypes.Structure):
            _fields_ = [('biSize', wintypes.DWORD), ('biWidth', wintypes.LONG), ('biHeight', wintypes.LONG),
                        ('biPlanes', wintypes.WORD), ('biBitCount', wintypes.WORD), ('biCompression', wintypes.DWORD),
                        ('biSizeImage', wintypes.DWORD), ('biXPelsPerMeter', wintypes.LONG),
                        ('biYPelsPerMeter', wintypes.LONG), ('biClrUsed', wintypes.DWORD), ('biClrImportant', wintypes.DWORD)]
        bmi = BMI(ctypes.sizeof(BMI), w, -h, 1, 32, 0, 0, 0, 0, 0, 0)
        buf = ctypes.create_string_buffer(w * h * 4)
        gdi32.GetDIBits(mem, bmp, 0, h, buf, ctypes.byref(bmi), 0)
        img = Image.frombuffer('RGBA', (w, h), buf, 'raw', 'BGRA', 0, 1).convert('RGB')
    gdi32.SelectObject(mem, old)
    gdi32.DeleteObject(bmp)
    gdi32.DeleteDC(mem)
    user32.ReleaseDC(hwnd, hdc)
    if img is not None and img.getbbox() is not None:
        return img
    # PrintWindow gave black: copy from the screen instead (needs the window uncovered)
    pt = wintypes.POINT(0, 0)
    user32.ClientToScreen(hwnd, ctypes.byref(pt))
    return ImageGrab.grab(bbox=(pt.x, pt.y, pt.x + w, pt.y + h)).convert('RGB')


def test_card(size):
    """A picture that shows orientation and size at a glance: colour bars, an arrow, a border."""
    w, h = size
    img = Image.new('RGB', (w, h), (0, 0, 0))
    d = ImageDraw.Draw(img)
    bars = [(255, 40, 60), (255, 150, 40), (250, 220, 60), (60, 220, 120), (60, 130, 255), (130, 90, 250), (230, 80, 255)]
    bw = w // len(bars)
    for i, c in enumerate(bars):
        d.rectangle([i * bw, h // 3, (i + 1) * bw - 2, h // 3 + h // 5], outline=c, width=3)
        for k in range(0, h // 5, 6):
            d.line([i * bw, h // 3 + k, (i + 1) * bw - 2, h // 3 + k + 6], fill=c)
    d.rectangle([0, 0, w - 1, h - 1], outline=(255, 255, 255), width=2)
    d.polygon([(w // 2, 10), (w // 2 - 22, 46), (w // 2 + 22, 46)], fill=(255, 255, 255))   # points to the top
    d.text((10, h - 40), 'LINGSYS  %dx%d' % (w, h), fill=(255, 255, 255))
    d.text((10, h - 24), 'TOP IS UP', fill=(180, 180, 180))
    return img


def main():
    sys.stdout.reconfigure(line_buffering=True)
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument('--title', default='LingSys Mini', help='title of the window to mirror')
    ap.add_argument('--fps', type=float, default=10.0)
    ap.add_argument('--rotate', type=int, default=90, choices=[0, 90, 180, 270],
                    help="clockwise turn applied to the window picture before sending (default 90: a 320 x 240 window fills the 240 x 320 panel)")
    ap.add_argument('--quality', type=int, default=85)
    ap.add_argument('--test', action='store_true', help='send a test card and exit')
    ap.add_argument('--save', help='also write the last encoded frame to this .jpg')
    a = ap.parse_args()

    dev = open_device()
    code, sub, serial = handshake(dev)
    size = PANELS.get(code, DEFAULT_PANEL)
    print('display: panel code %d (sub %d), serial %s, frames %dx%d%s' %
          (code, sub, serial, size[0], size[1],
           '' if code in PANELS else '  [code not in my table, using the default size]'))

    if a.test:
        jpeg = encode(test_card((size[1], size[0]) if a.rotate in (90, 270) else size), size, a.rotate, a.quality)
        send_frame(dev, make_packet(jpeg, size))
        if a.save:
            open(a.save, 'wb').write(jpeg)
        print('test card sent (%d bytes of JPEG). Top of the card should be at the top of the display.' % len(jpeg))
        return

    period = 1.0 / a.fps
    sent, t0, last_msg = 0, time.time(), 0.0
    while True:
        t = time.time()
        hwnd = find_window(a.title)
        img = grab_window(hwnd) if hwnd else None
        if img is None:
            if t - last_msg > 3:
                print('waiting for a window titled "%s"...' % a.title)
                last_msg = t
            time.sleep(0.5)
            continue
        jpeg = encode(img, size, a.rotate, a.quality)
        try:
            send_frame(dev, make_packet(jpeg, size))
        except usb.core.USBError as e:
            print('USB error (%s); reconnecting' % e)
            time.sleep(1.0)
            dev = open_device()
            handshake(dev)
            continue
        sent += 1
        if a.save:
            open(a.save, 'wb').write(jpeg)
        if sent % 100 == 0:
            print('%d frames, %.1f fps, last frame %d bytes' % (sent, sent / (time.time() - t0), len(jpeg)))
        dt = time.time() - t
        if dt < period:
            time.sleep(period - dt)


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        pass