112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
import t, c
|
|
import drivers.usb.usb as usb
|
|
import drivers.usb.hid.hid as hid
|
|
import drivers.usb.uhci as uhci
|
|
import drivers.serial.uart.serial as serial
|
|
|
|
KBD_BUF_SIZE: t.CDefine = 1024
|
|
KBD_EVT_PRESS: t.CDefine = 0
|
|
KBD_EVT_RELEASE: t.CDefine = 1
|
|
|
|
class usb_kbd_event(t.CStruct):
|
|
modifier: t.CUInt8T
|
|
keycode: t.CUInt8T
|
|
event_type: t.CUInt8T
|
|
|
|
kbd_buf: t.CArray[usb_kbd_event, KBD_BUF_SIZE]
|
|
kbd_buf_head: t.CInt = 0
|
|
kbd_buf_tail: t.CInt = 0
|
|
kbd_buf_count: t.CInt = 0
|
|
kbd_dev: usb.usb_device | t.CPtr = None
|
|
kbd_report: hid.hid_kbd_report
|
|
kbd_prev_keycodes: t.CArray[t.CUInt8T, 6]
|
|
kbd_initialized: t.CInt = 0
|
|
kbd_slot_id: t.CInt = -1
|
|
|
|
def init() -> t.CInt:
|
|
global kbd_dev, kbd_initialized, kbd_slot_id
|
|
dev: usb.usb_device | t.CPtr = usb.find_hid_device(0x01)
|
|
if dev is None:
|
|
serial.puts("[usb_kbd] no USB keyboard found\n")
|
|
return -1
|
|
kbd_dev = dev
|
|
if hid.init_keyboard(dev) != 0:
|
|
serial.puts("[usb_kbd] init failed\n")
|
|
return -2
|
|
kbd_initialized = 1
|
|
serial.puts("[usb_kbd] initialized\n")
|
|
kbd_slot_id = uhci.register_int_callback(c.Addr(_on_int_complete))
|
|
if kbd_slot_id < 0:
|
|
serial.puts("[usb_kbd] register callback failed\n")
|
|
kbd_initialized = 0
|
|
return -3
|
|
uhci.schedule_int_transfer(kbd_slot_id, dev.addr, dev.ep_in, dev.ls, t.CUInt8T(dev.ep_in_max), dev.ep_in_toggle, c.Addr(kbd_report))
|
|
return 0
|
|
|
|
def _on_int_complete() -> t.CInt:
|
|
global kbd_buf_head, kbd_buf_tail, kbd_buf_count, kbd_prev_keycodes
|
|
if kbd_initialized == 0: return 0
|
|
i: t.CInt
|
|
j: t.CInt
|
|
for i in range(6):
|
|
code: t.CUInt8T = kbd_report.keycode[i]
|
|
if code == 0: continue
|
|
found: t.CInt = 0
|
|
for j in range(6):
|
|
if code == kbd_prev_keycodes[j]:
|
|
found = 1
|
|
break
|
|
if found == 0:
|
|
if kbd_buf_count < KBD_BUF_SIZE:
|
|
kbd_buf[kbd_buf_tail].modifier = kbd_report.modifier
|
|
kbd_buf[kbd_buf_tail].keycode = code
|
|
kbd_buf[kbd_buf_tail].event_type = KBD_EVT_PRESS
|
|
kbd_buf_tail = kbd_buf_tail + 1
|
|
if kbd_buf_tail >= KBD_BUF_SIZE:
|
|
kbd_buf_tail = 0
|
|
kbd_buf_count += 1
|
|
for i in range(6):
|
|
prev: t.CUInt8T = kbd_prev_keycodes[i]
|
|
if prev == 0: continue
|
|
found2: t.CInt = 0
|
|
for j in range(6):
|
|
if prev == kbd_report.keycode[j]:
|
|
found2 = 1
|
|
break
|
|
if found2 == 0:
|
|
if kbd_buf_count < KBD_BUF_SIZE:
|
|
kbd_buf[kbd_buf_tail].modifier = kbd_report.modifier
|
|
kbd_buf[kbd_buf_tail].keycode = prev
|
|
kbd_buf[kbd_buf_tail].event_type = KBD_EVT_RELEASE
|
|
kbd_buf_tail = kbd_buf_tail + 1
|
|
if kbd_buf_tail >= KBD_BUF_SIZE:
|
|
kbd_buf_tail = 0
|
|
kbd_buf_count += 1
|
|
for i in range(6):
|
|
kbd_prev_keycodes[i] = kbd_report.keycode[i]
|
|
dev: usb.usb_device | t.CPtr = kbd_dev
|
|
if dev is not None and kbd_slot_id >= 0:
|
|
uhci.schedule_int_transfer(kbd_slot_id, dev.addr, dev.ep_in, dev.ls, t.CUInt8T(dev.ep_in_max), dev.ep_in_toggle, c.Addr(kbd_report))
|
|
return 0
|
|
|
|
def has_data() -> t.CInt:
|
|
if kbd_buf_count > 0: return 1
|
|
return 0
|
|
|
|
def poll() -> t.CInt:
|
|
if kbd_initialized == 0: return 0
|
|
if kbd_buf_count > 0: return 1
|
|
return 0
|
|
|
|
def read_event(out: usb_kbd_event | t.CPtr) -> t.CInt:
|
|
global kbd_buf_head, kbd_buf_count
|
|
if kbd_buf_count == 0: return -1
|
|
out.modifier = kbd_buf[kbd_buf_head].modifier
|
|
out.keycode = kbd_buf[kbd_buf_head].keycode
|
|
out.event_type = kbd_buf[kbd_buf_head].event_type
|
|
kbd_buf_head = kbd_buf_head + 1
|
|
if kbd_buf_head >= KBD_BUF_SIZE:
|
|
kbd_buf_head = 0
|
|
kbd_buf_count -= 1
|
|
return 0
|