291 lines
10 KiB
Python
291 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Local Kodi EventServer (Yatse UDP 9777) test utility and CLI.
|
|
|
|
This script allows you to test the Kodi EventServer locally by sending
|
|
valid Kodi EventServer UDP packets (HELO, PING, BUTTON, ACTION) and
|
|
verifying handshake responses.
|
|
|
|
Usage examples:
|
|
# 1. Test HELO and PING handshake
|
|
python3 tests/test_eventserver_client.py ping
|
|
|
|
# 2. Send navigation buttons
|
|
python3 tests/test_eventserver_client.py up
|
|
python3 tests/test_eventserver_client.py down
|
|
python3 tests/test_eventserver_client.py left
|
|
python3 tests/test_eventserver_client.py right
|
|
python3 tests/test_eventserver_client.py select
|
|
python3 tests/test_eventserver_client.py back
|
|
|
|
# 3. Interactive keyboard control mode:
|
|
python3 tests/test_eventserver_client.py interactive
|
|
|
|
# 4. Run automated test suite against running server:
|
|
python3 tests/test_eventserver_client.py test
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import random
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
# Kodi EventServer Constants
|
|
HEADER_FORMAT = "!4sBBHIIHI10s"
|
|
HEADER_SIZE = struct.calcsize(HEADER_FORMAT) # 32 bytes
|
|
|
|
PT_HELO = 0x01
|
|
PT_BYE = 0x02
|
|
PT_BUTTON = 0x03
|
|
PT_PING = 0x05
|
|
PT_ACTION = 0x0A
|
|
|
|
PTB_DOWN = 0x02
|
|
PTB_UP = 0x04
|
|
|
|
|
|
class EventServerClient:
|
|
"""UDP Client simulating Yatse / Kodi remote."""
|
|
|
|
def __init__(self, host: str = "127.0.0.1", port: int = 9777, timeout: float = 2.0):
|
|
self.host = host
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self.token = random.randint(100000, 999999)
|
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.sock.settimeout(timeout)
|
|
|
|
def close(self) -> None:
|
|
self.sock.close()
|
|
|
|
def _build_header(self, pkttype: int, paylen: int) -> bytes:
|
|
return struct.pack(
|
|
HEADER_FORMAT,
|
|
b"XBMC", # Signature
|
|
2, # Major version
|
|
0, # Minor version
|
|
pkttype, # Packet type
|
|
1, # Seq
|
|
1, # MaxSeq
|
|
paylen, # Payload length
|
|
self.token,
|
|
b"\x00" * 10,
|
|
)
|
|
|
|
def send_helo(self, device_name: str = "Local Yatse Test") -> bool:
|
|
"""Send HELO packet and wait for PING ACK response."""
|
|
payload = (
|
|
device_name.encode("utf-8") + b"\x00" +
|
|
b"\x00" + # Icon type
|
|
struct.pack("!H", 0) + # Port
|
|
b"\x00" * 8 # Reserved
|
|
)
|
|
packet = self._build_header(PT_HELO, len(payload)) + payload
|
|
self.sock.sendto(packet, (self.host, self.port))
|
|
|
|
try:
|
|
data, addr = self.sock.recvfrom(1024)
|
|
if len(data) >= HEADER_SIZE:
|
|
sig, maj, minv, pkttype, seq, maxseq, plen, token, _ = struct.unpack(
|
|
HEADER_FORMAT, data[:HEADER_SIZE]
|
|
)
|
|
if sig == b"XBMC" and pkttype == PT_PING and token == self.token:
|
|
return True
|
|
except socket.timeout:
|
|
return False
|
|
return False
|
|
|
|
def send_ping(self) -> bool:
|
|
"""Send PING packet and wait for PING ACK response."""
|
|
packet = self._build_header(PT_PING, 0)
|
|
self.sock.sendto(packet, (self.host, self.port))
|
|
|
|
try:
|
|
data, addr = self.sock.recvfrom(1024)
|
|
if len(data) >= HEADER_SIZE:
|
|
sig, _, _, pkttype, _, _, _, token, _ = struct.unpack(
|
|
HEADER_FORMAT, data[:HEADER_SIZE]
|
|
)
|
|
if sig == b"XBMC" and pkttype == PT_PING and token == self.token:
|
|
return True
|
|
except socket.timeout:
|
|
return False
|
|
return False
|
|
|
|
def send_button(self, button_name: str, map_name: str = "R1", is_down: bool = True) -> None:
|
|
"""Send BUTTON packet (e.g. up, down, left, right, select, back)."""
|
|
flags = PTB_DOWN if is_down else PTB_UP
|
|
payload = (
|
|
struct.pack("!HHH", 0, flags, 0) +
|
|
map_name.encode("utf-8") + b"\x00" +
|
|
button_name.encode("utf-8") + b"\x00"
|
|
)
|
|
packet = self._build_header(PT_BUTTON, len(payload)) + payload
|
|
self.sock.sendto(packet, (self.host, self.port))
|
|
|
|
def send_action(self, action_str: str) -> None:
|
|
"""Send ACTION packet (e.g. Action(1), Action(7), Up)."""
|
|
payload = b"\x01" + action_str.encode("utf-8") + b"\x00"
|
|
packet = self._build_header(PT_ACTION, len(payload)) + payload
|
|
self.sock.sendto(packet, (self.host, self.port))
|
|
|
|
def send_bye(self) -> None:
|
|
"""Send BYE disconnect packet."""
|
|
packet = self._build_header(PT_BYE, 0)
|
|
self.sock.sendto(packet, (self.host, self.port))
|
|
|
|
|
|
def run_automated_tests(host: str, port: int) -> bool:
|
|
print(f"Testing Kodi EventServer at {host}:{port}...")
|
|
client = EventServerClient(host, port, timeout=2.0)
|
|
|
|
# Test 1: HELO Handshake
|
|
print(" [1/4] Testing HELO handshake (client registration)...", end=" ", flush=True)
|
|
if client.send_helo("EventServer Test Suite"):
|
|
print("OK (Received PING-ACK)")
|
|
else:
|
|
print("FAILED (No PING-ACK response received)")
|
|
print(f" Ensure jellyfin-kiosk is running and listening on UDP {port}")
|
|
client.close()
|
|
return False
|
|
|
|
# Test 2: PING Heartbeat
|
|
print(" [2/4] Testing PING heartbeat...", end=" ", flush=True)
|
|
if client.send_ping():
|
|
print("OK (Received PING-ACK)")
|
|
else:
|
|
print("FAILED")
|
|
client.close()
|
|
return False
|
|
|
|
# Test 3: Button Events
|
|
print(" [3/4] Sending navigation buttons (Up, Down, Left, Right, Select, Back)...", end=" ", flush=True)
|
|
for btn in ["up", "down", "left", "right", "select", "back"]:
|
|
client.send_button(btn)
|
|
time.sleep(0.05)
|
|
print("OK (Packets sent)")
|
|
|
|
# Test 4: Action Events
|
|
print(" [4/4] Sending action strings (Action(1), Action(7))...", end=" ", flush=True)
|
|
for act in ["Action(1)", "Action(7)"]:
|
|
client.send_action(act)
|
|
time.sleep(0.05)
|
|
print("OK (Packets sent)")
|
|
|
|
client.send_bye()
|
|
client.close()
|
|
print("\nAll EventServer tests passed successfully!")
|
|
return True
|
|
|
|
|
|
def run_interactive_mode(client: EventServerClient) -> None:
|
|
print("\n--- EventServer Interactive Remote Mode ---")
|
|
print("Controls:")
|
|
print(" [W] or [Up] : Up")
|
|
print(" [S] or [Down] : Down")
|
|
print(" [A] or [Left] : Left")
|
|
print(" [D] or [Right] : Right")
|
|
print(" [Enter] / [O] : Select / OK")
|
|
print(" [B] / [Esc] : Back")
|
|
print(" [Space] / [P] : Play / Pause")
|
|
print(" [X] : Stop")
|
|
print(" [Q] : Quit\n")
|
|
|
|
import termios
|
|
import tty
|
|
|
|
fd = sys.stdin.fileno()
|
|
old_settings = termios.tcgetattr(fd)
|
|
try:
|
|
tty.setraw(fd)
|
|
while True:
|
|
ch = sys.stdin.read(1)
|
|
if ch == "\x1b": # Escape or arrow sequence
|
|
# Non-blocking check for arrow keys
|
|
import select
|
|
r, _, _ = select.select([sys.stdin], [], [], 0.05)
|
|
if r:
|
|
seq = sys.stdin.read(2)
|
|
if seq == "[A":
|
|
client.send_button("up")
|
|
elif seq == "[B":
|
|
client.send_button("down")
|
|
elif seq == "[C":
|
|
client.send_button("right")
|
|
elif seq == "[D":
|
|
client.send_button("left")
|
|
else:
|
|
client.send_button("back")
|
|
elif ch in ("q", "Q", "\x03"): # Quit or Ctrl+C
|
|
break
|
|
elif ch in ("w", "W"):
|
|
client.send_button("up")
|
|
elif ch in ("s", "S"):
|
|
client.send_button("down")
|
|
elif ch in ("a", "A"):
|
|
client.send_button("left")
|
|
elif ch in ("d", "D"):
|
|
client.send_button("right")
|
|
elif ch in ("\r", "\n", "o", "O"):
|
|
client.send_button("select")
|
|
elif ch in ("b", "B"):
|
|
client.send_button("back")
|
|
elif ch in (" ", "p", "P"):
|
|
client.send_button("playpause")
|
|
elif ch in ("x", "X"):
|
|
client.send_button("stop")
|
|
finally:
|
|
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
|
print("\nExited interactive mode.")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Kodi EventServer Local Test Client")
|
|
parser.add_argument("command", nargs="?", default="test",
|
|
help="Command: test, ping, helo, up, down, left, right, select, back, play, stop, interactive")
|
|
parser.add_argument("--host", default="127.0.0.1", help="EventServer host (default: 127.0.0.1)")
|
|
parser.add_argument("--port", type=int, default=9777, help="EventServer UDP port (default: 9777)")
|
|
|
|
args = parser.parse_args()
|
|
cmd = args.command.lower()
|
|
|
|
if cmd == "test":
|
|
success = run_automated_tests(args.host, args.port)
|
|
return 0 if success else 1
|
|
|
|
client = EventServerClient(args.host, args.port)
|
|
try:
|
|
if cmd == "ping":
|
|
print(f"Sending PING to {args.host}:{args.port}...")
|
|
if client.send_ping():
|
|
print("Success: Received PING-ACK from EventServer!")
|
|
else:
|
|
print("Failed: No response (timeout)")
|
|
return 1
|
|
elif cmd == "helo":
|
|
print(f"Sending HELO to {args.host}:{args.port}...")
|
|
if client.send_helo():
|
|
print("Success: EventServer accepted HELO and replied with PING-ACK!")
|
|
else:
|
|
print("Failed: No response (timeout)")
|
|
return 1
|
|
elif cmd in ("up", "down", "left", "right", "select", "back", "play", "pause", "playpause", "stop"):
|
|
client.send_button(cmd)
|
|
print(f"Sent button '{cmd}' to {args.host}:{args.port}")
|
|
elif cmd == "interactive":
|
|
run_interactive_mode(client)
|
|
else:
|
|
print(f"Unknown command: {cmd}")
|
|
return 1
|
|
finally:
|
|
client.close()
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|