hudiy.eu

Code: Select all

def trigger_action(self, client, action):
    dispatch_action = hudiy_api.DispatchAction()
    dispatch_action.action = action
    client.send(hudiy_api.MESSAGE_DISPATCH_ACTION, 0,
                dispatch_action.SerializeToString())
                
self.trigger_action(client, "output_volume_up")
self.trigger_action(client, "output_volume_down")
self.trigger_action(client, "toggle_output_muted")

Code: Select all

import smbus
from pynput.keyboard import Key, Controller
import lgpio
import time
import os
import threading

# ==================================================
# I2C constants
# ==================================================
I2C_BUS = 1
I2C_ADDRESS = 0x08

# Messages
ALIVE_MSG       = 0x4C   # Pi -> Teensy
COUNTDOWN_MSG   = 0x43   # Pi -> Teensy (countdown active)
SHUTDOWN_REQ    = 0x3B   # Teensy -> Pi (shutdown request)
SHUTDOWN_ACK    = 0x53   # Pi -> Teensy
SHUTTING_DOWN   = 0x44   # Pi -> Teensy ("going down")

# Key control signals
NO_KEY = 0x00
HARDBOOT_SIGNAL = 0x52
MCU_RESET_SIGNAL = 0x54
GPIO13 = 13

# HDMI control
HDMI_OFF_DELAY = 10  # seconds before HDMI goes off in countdown

# ==================================================
# Key mappings
# ==================================================
KEYS = {
    0x00: None,
    0xCC: Key.f11,    # Mute key
    0xC9: Key.f8,     # VolCW
    0xC8: Key.f7,     # VolCCW
    0xDA: Key.up,
    0xD9: Key.down,
    0xD8: Key.left,
    0xD7: Key.right,
    0xB0: Key.enter,
    0xB1: Key.esc,
    0x48: 'H',
    0x4A: 'J',
    0x46: 'F',
    0xC4: Key.f3,     # Fav (with Ctrl pressed)
    0x31: '1',
    0x32: '2',
    0x52: 'R',
    0x54: 'T',
    0x4C: 'L',
    0x4D: 'M',
    0x70: 'P',
    0x6F: 'O',
    0x6E: 'N',
    0x76: 'V',
}

# ==================================================
# Globals
# ==================================================
keyboard = Controller()
bus = smbus.SMBus(I2C_BUS)

# GPIO setup (lgpio)
CHIP = 0
chip = lgpio.gpiochip_open(CHIP)
lgpio.gpio_claim_output(chip, GPIO13)
lgpio.gpio_write(chip, GPIO13, 1)  # set HIGH initially

running = True
shutdown_initiated = False
in_countdown = False

# HDMI state
hdmi_active = True
hdmi_timer = 0
hdmi_lock = threading.Lock()

# ==================================================
# Helpers
# ==================================================
def press_key_with_delay(key, delay=0.05):
    keyboard.press(key)
    time.sleep(delay)
    keyboard.release(key)

def execute_keystroke(key_code):
    key = KEYS.get(key_code)
    if key is None:
        return
    if key_code == 0xC4:  # Ctrl + F3
        with keyboard.pressed(Key.ctrl):
            press_key_with_delay(Key.f3)
    elif isinstance(key, Key):
        press_key_with_delay(key)
    elif isinstance(key, str):
        keyboard.type(key)

# ==================================================
# HDMI Control (Bookworm / Pi 5)
# ==================================================
def hdmi_on():
    global hdmi_active
    with hdmi_lock:
        if not hdmi_active:
            print("[HDMI] Screen will restore on keypress")
            hdmi_active = True

def hdmi_off():
    global hdmi_active
    with hdmi_lock:
        if hdmi_active:
            print("[HDMI] Blanking screen (any key will restore)")
            os.system("kmsblank")
            hdmi_active = False

def hdmi_countdown_worker():
    global hdmi_timer
    while running:
        if in_countdown:
            if hdmi_timer > 0:
                print(f"[HDMI] Countdown: {hdmi_timer}s")
                time.sleep(1)
                hdmi_timer -= 1
            elif hdmi_active:
                hdmi_off()
                time.sleep(1)
        else:
            time.sleep(1)

# ==================================================
# Key / Control handling
# ==================================================
def process_key_event(key_byte):
    if key_byte == HARDBOOT_SIGNAL:
        print("[Pi] Hardboot signal received, rebooting.")
        os.system("sudo reboot")
    elif key_byte == MCU_RESET_SIGNAL:
        print("[Pi] MCU reset signal received, toggling GPIO13.")
        lgpio.gpio_write(chip, GPIO13, 0)
        time.sleep(0.2)
        lgpio.gpio_write(chip, GPIO13, 1)
    else:
        print(f"[Pi] Key Received: {hex(key_byte)}")
        execute_keystroke(key_byte)

# ==================================================
# Heartbeat Task
# ==================================================
def heartbeat_task():
    global running, shutdown_initiated, in_countdown
    while running and not shutdown_initiated:
        try:
            if in_countdown:
                bus.write_byte(I2C_ADDRESS, COUNTDOWN_MSG)
                print("[Pi] Sent COUNTDOWN (still alive, preparing).")
            else:
                bus.write_byte(I2C_ADDRESS, ALIVE_MSG)
                print("[Pi] Sent ALIVE.")
        except Exception as e:
            print(f"[Pi] I2C write error: {e}")
        time.sleep(5)

# ==================================================
# Listen for Teensy messages
# ==================================================
def listen_for_teensy():
    global shutdown_initiated, in_countdown, hdmi_timer
    while running and not shutdown_initiated:
        try:
            msg = bus.read_byte(I2C_ADDRESS)

            if msg == SHUTDOWN_REQ:
                print("[Pi] Teensy requested shutdown.")
                try:
                    bus.write_byte(I2C_ADDRESS, SHUTDOWN_ACK)
                    print("[Pi] Sent SHUTDOWN ACK to Teensy.")
                except Exception as e:
                    print(f"[Pi] I2C write error (ACK): {e}")

                try:
                    bus.write_byte(I2C_ADDRESS, SHUTTING_DOWN)
                    print("[Pi] Sent SHUTTING DOWN signal.")
                except Exception as e:
                    print(f"[Pi] I2C write error (shutdown msg): {e}")

                shutdown_initiated = True
                print("[Pi] Executing shutdown now.")
                os.system("sudo shutdown now")

            elif msg == COUNTDOWN_MSG:
                if not in_countdown:
                    print("[Pi] Teensy entered COUNTDOWN mode.")
                    hdmi_timer = HDMI_OFF_DELAY
                in_countdown = True

            elif msg == NO_KEY:
                if in_countdown:
                    print("[Pi] Countdown cancelled, back to ALIVE mode.")
                    hdmi_on()
                in_countdown = False

            else:
                process_key_event(msg)

        except Exception:
            time.sleep(1)

# ==================================================
# Main
# ==================================================
if __name__ == "__main__":
    t1 = threading.Thread(target=heartbeat_task, daemon=True)
    t2 = threading.Thread(target=listen_for_teensy, daemon=True)
    t3 = threading.Thread(target=hdmi_countdown_worker, daemon=True)
    t1.start()
    t2.start()
    t3.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        running = False
        print("[Pi] Stopped by user.")
    finally:
        # Cleanup GPIO
        lgpio.gpiochip_close(chip)

Code: Select all

async def import_uinput():
    global events, device, uinput
    try:
        events = (
            uinput.KEY_1, uinput.KEY_2, uinput.KEY_UP, uinput.KEY_DOWN, uinput.KEY_LEFT, uinput.KEY_RIGHT,
            uinput.KEY_ENTER, uinput.KEY_ESC, uinput.KEY_F2, uinput.KEY_B, uinput.KEY_N, uinput.KEY_V,
            uinput.KEY_F12, uinput.KEY_M, uinput.KEY_X, uinput.KEY_C, uinput.KEY_LEFTCTRL, uinput.KEY_H, uinput.KEY_T
        )
        device = uinput.Device(events)

Code: Select all

        elif msg == '373004000200' and back > 0:  # RNS-E: return button released
            if back <= 4:
                device.emit(uinput.KEY_ESC, 1)
                device.emit(uinput.KEY_ESC, 0)
                back = 0