USB キーボード信号の送受信とマウス操作｜Raspberry Pi Zero 2 W を USB マウスに変える（記事第2弾）

nano keyboard-rx.py

import socket
import json
import sys
import termios
import tty

HOST = '0.0.0.0'
PORT = 65432

# HIDファイルのパス
MOUSE_DEV = "/dev/hidg0"

# 初期設定
movement_amount = 10
min_movement = 1
max_movement = 127

# キーボード入力を非同期で取得するための設定
def get_key():
    fd = sys.stdin.fileno()
    old_settings = termios.tcgetattr(fd)
    try:
        tty.setraw(sys.stdin.fileno())
        ch = sys.stdin.read(1)
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
    return ch

# マウスの移動やクリックを送信する関数
def move_mouse(dx, dy, buttons=0):
    report = bytearray([buttons, dx & 0xFF, dy & 0xFF])
    with open(MOUSE_DEV, "wb") as f:
        f.write(report)

# ソケットサーバーでの接続待機
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    print('Waiting for a connection...')

    conn, addr = s.accept()
    with conn:
        print(f'Connected by {addr}')
        buffer = ""
        while True:
            data = conn.recv(1024)
            if not data:
                break
            buffer += data.decode('utf-8')
            while True:
                try:
                    # JSONオブジェクトの境界を探す
                    obj, index = json.JSONDecoder().raw_decode(buffer)
                    buffer = buffer[index:].lstrip()  # バッファから処理済み部分を削除

                    if obj['type'] == 1:  # EV_KEY
                        if obj['value'] in (1, 2):  # キーが押された、または押し続けられている
                            if obj['code'] == 103:  # 上方向キー
                                move_mouse(0, -movement_amount)
                            elif obj['code'] == 108:  # 下方向キー
                                move_mouse(0, movement_amount)
                            elif obj['code'] == 105:  # 左方向キー
                                move_mouse(-movement_amount, 0)
                            elif obj['code'] == 106:  # 右方向キー
                                move_mouse(movement_amount, 0)
                            elif obj['code'] == 44:  # Zキー
                                movement_amount = max(min_movement, movement_amount - 4)
                            elif obj['code'] == 45:  # Xキー
                                movement_amount = min(max_movement, movement_amount + 4)
                            elif obj['code'] == 30:  # Aキーで左クリック
                                move_mouse(0, 0, buttons=1)
                            elif obj['code'] == 31:  # Sキーで右クリック
                                move_mouse(0, 0, buttons=2)
                            elif obj['code'] == 16:  # Qキー
                                print("Q key pressed. Exiting...")
                                conn.close()
                                s.close()
                                exit()
                        elif obj['value'] == 0:  # キーが離されたとき、クリックを解除
                            if obj['code'] in (30, 31):  # AキーまたはSキー
                                move_mouse(0, 0, buttons=0)

                    print(f"Received: {obj}")

                except json.JSONDecodeError:
                    break

print("終了しました。")

sudo python3 keyboard-rx.py

Waiting for a connection...

nano keyboard-tx.py

import evdev
import socket
import json

# 送信先のホスト名とポート番号
HOST = 'zero-2.local'  # 受信側のRaspberry Piのホスト名またはIPアドレス
PORT = 65432

# デバイスのevent番号を指定（キーボードのイベントデバイス）
device = evdev.InputDevice('/dev/input/event0')

# ソケットを作成し接続
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    try:
        s.connect((HOST, PORT))
        print(f"Connected to {HOST}:{PORT}")
        # キーボードイベントの読み取りと送信
        for event in device.read_loop():
            if event.type == evdev.ecodes.EV_KEY:
                # データの送信 (JSON + 改行)
                data = json.dumps({'type': event.type, 'code': event.code, 'value': event.value})
                s.sendall((data + '\n').encode('utf-8'))
                print(f"Sent: {data}")  # 送信したデータを表示
    except BrokenPipeError:
        print("Connection lost. Exiting.")
    except Exception as e:
        print(f"An error occurred: {e}")

nano keyboard-tx.py