Raspyjack/evdev_keys.py at main · 7h30th3r0n3/Raspyjack · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
"""
evdev_keys – TCA8418 keyboard reader for M5Stack CardputerZero
"""

import threading
import evdev
import os

# Auto-detect TCA8418 keyboard device
EVDEV_DEVICE = os.environ.get('RJ_KEYBOARD_DEVICE', '')
if not EVDEV_DEVICE:
    EVDEV_DEVICE = '/dev/input/event3'  # fallback
    for _i in range(8):
        try:
            with open(f'/sys/class/input/event{_i}/device/name') as _n:
                if 'tca8418' in _n.read().lower():
                    EVDEV_DEVICE = f'/dev/input/event{_i}'
                    break
        except Exception:
            pass

_KEYMAP = {
    103: 'KEY_UP_PIN',
    108: 'KEY_DOWN_PIN',
    105: 'KEY_LEFT_PIN',
    106: 'KEY_RIGHT_PIN',
    33:  'KEY_UP_PIN',
    44:  'KEY_LEFT_PIN',
    45:  'KEY_DOWN_PIN',
    46:  'KEY_RIGHT_PIN',
    28:  'KEY_PRESS_PIN',
    1:   'KEY3_PIN',
    57:  'KEY1_PIN',
    14:  'KEY2_PIN',
    15:  'KEY3_PIN',
}

_REVERSE_MAP = {}
for _code, _name in _KEYMAP.items():
    _REVERSE_MAP.setdefault(_name, []).append(_code)

_key_state = {}
_lock = threading.Lock()
_device = None
_thread = None
_RAW_PIPE = "/dev/shm/rj_key_events"


_PAUSE_FLAG = "/dev/shm/rj_evdev_pause"


def _reader_loop():
    global _device
    while True:
        try:
            if os.path.exists(_PAUSE_FLAG):
                if _device:
                    try:
                        _device.close()
                    except Exception:
                        pass
                    _device = None
                import time
                time.sleep(0.2)
                continue
            if _device is None:
                _device = evdev.InputDevice(EVDEV_DEVICE)
            for event in _device.read_loop():
                if os.path.exists(_PAUSE_FLAG):
                    try:
                        _device.close()
                    except Exception:
                        pass
                    _device = None
                    break
                if event.type != evdev.ecodes.EV_KEY:
                    continue
                with _lock:
                    _key_state[event.code] = event.value > 0
        except Exception:
            _device = None
            import time
            time.sleep(0.5)


def start():
    global _thread
    if _thread is not None:
        return
    _thread = threading.Thread(target=_reader_loop, daemon=True)
    _thread.start()


def is_pressed(button_name: str) -> bool:
    codes = _REVERSE_MAP.get(button_name, [])
    with _lock:
        return any(_key_state.get(c, False) for c in codes)


def is_key_pressed(evdev_code: int) -> bool:
    with _lock:
        return _key_state.get(evdev_code, False)


def any_pressed() -> bool:
    with _lock:
        return any(_key_state.values())


def get_pressed_button():
    with _lock:
        for code, pressed in _key_state.items():
            if pressed and code in _KEYMAP:
                return _KEYMAP[code]
    return None


def pause():
    try:
        with open(_PAUSE_FLAG, "w") as f:
            pass
    except Exception:
        pass
    import time
    time.sleep(0.3)


def resume():
    try:
        os.unlink(_PAUSE_FLAG)
    except OSError:
        pass


if not os.path.exists(_PAUSE_FLAG):
    start()