joycon_test.py

import os
import sys
import importlib
import math
import time
import traceback
import errno
import argparse


PROJECT_ROOT = os.path.dirname(__file__)
PYJOYCON_ROOT = os.path.join(PROJECT_ROOT, "joycon-python")
if PYJOYCON_ROOT not in sys.path:
	sys.path.insert(0, PYJOYCON_ROOT)

pyjoycon = importlib.import_module("pyjoycon")
constants = importlib.import_module("pyjoycon.constants")
evdev_hid = importlib.import_module("pyjoycon.evdev_hid")

JoyCon = pyjoycon.JoyCon
GyroTrackingJoyCon = pyjoycon.GyroTrackingJoyCon
ButtonEventJoyCon = pyjoycon.ButtonEventJoyCon
get_L_id = pyjoycon.get_L_id
get_R_id = pyjoycon.get_R_id

JOYCON_L_PRODUCT_ID = constants.JOYCON_L_PRODUCT_ID
JOYCON_PRODUCT_IDS = constants.JOYCON_PRODUCT_IDS
JOYCON_R_PRODUCT_ID = constants.JOYCON_R_PRODUCT_ID
JOYCON_VENDOR_ID = constants.JOYCON_VENDOR_ID

EVDEV_HIDRAW_SERIAL_PREFIX = evdev_hid.EVDEV_HIDRAW_SERIAL_PREFIX

RIGHT_BUTTON_PROMPTS = [
	("A", "a"),
	("B", "b"),
	("X", "x"),
	("Y", "y"),
	("R", "r"),
	("ZR", "zr"),
	("PLUS", "plus"),
	("HOME", "home"),
	("RIGHT_SR", "right_sr"),
	("RIGHT_SL", "right_sl"),
	("STICK_R_BTN", "stick_r_btn"),
]

LEFT_BUTTON_PROMPTS = [
	("UP", "up"),
	("DOWN", "down"),
	("LEFT", "left"),
	("RIGHT", "right"),
	("L", "l"),
	("ZL", "zl"),
	("MINUS", "minus"),
	("CAPTURE", "capture"),
	("LEFT_SR", "left_sr"),
	("LEFT_SL", "left_sl"),
	("STICK_L_BTN", "stick_l_btn"),
]


class ReadmeCombinedJoyCon(GyroTrackingJoyCon, ButtonEventJoyCon):
	pass


def _is_valid_joycon_id(joycon_id):
	if not isinstance(joycon_id, tuple) or len(joycon_id) != 3:
		return False

	vendor_id, product_id, _serial = joycon_id
	return vendor_id == JOYCON_VENDOR_ID and product_id in JOYCON_PRODUCT_IDS


def _get_any_joycon_id():
	joycon_id = get_R_id(backend="evdev")
	if _is_valid_joycon_id(joycon_id):
		return joycon_id

	joycon_id = get_L_id(backend="evdev")
	if _is_valid_joycon_id(joycon_id):
		return joycon_id

	return (None, None, None)


def _get_side_joycon_id(side):
	if side == "R":
		return get_R_id(backend="evdev")
	if side == "L":
		return get_L_id(backend="evdev")
	raise ValueError(f"unsupported side: {side!r}")


def _assert_finite_triplet(values):
	assert isinstance(values, tuple)
	assert len(values) == 3
	for value in values:
		assert isinstance(value, (int, float))
		assert math.isfinite(value)


def _collect_joycon_permission_rows():
	rows = []
	try:
		from evdev import InputDevice, list_devices
	except ImportError:
		return rows

	for event_path in list_devices():
		dev = None
		try:
			dev = InputDevice(event_path)
			role = evdev_hid._classify_evdev_name(dev.name or "")
			if role is None:
				continue
			side, _is_imu = role
			hidraw_path = evdev_hid._event_to_hidraw_path(event_path)
			if not hidraw_path:
				continue

			rows.append({
				"side": side,
				"event": event_path,
				"hidraw": hidraw_path,
				"rw": os.access(hidraw_path, os.R_OK | os.W_OK),
			})
		except OSError:
			continue
		finally:
			if dev is not None:
				dev.close()

	# De-duplicate by event/hidraw pair.
	uniq = {}
	for row in rows:
		uniq[(row["event"], row["hidraw"])] = row
	return list(uniq.values())


def _print_joycon_permission_diagnostics():
	rows = _collect_joycon_permission_rows()
	if not rows:
		print("[DIAG] No JoyCon hidraw mapping discovered via evdev backend")
		return rows

	print("[DIAG] JoyCon hidraw access:")
	for row in rows:
		status = "rw-ok" if row["rw"] else "rw-denied"
		print(
			f"  - side={row['side']} event={row['event']} "
			f"hidraw={row['hidraw']} access={status}"
		)
	return rows


def _is_permission_error(exc):
	cur = exc
	while cur is not None:
		if isinstance(cur, PermissionError):
			return True
		if isinstance(cur, OSError) and getattr(cur, "errno", None) == errno.EACCES:
			return True
		cur = getattr(cur, "__cause__", None)
	return False


def test_evdev_backend_can_read_status():
	"""
	Integration test: find a JoyCon through evdev mapping and read one status frame.
	Requires a paired JoyCon and Linux input permissions.
	"""
	joycon_id = _get_any_joycon_id()

	assert _is_valid_joycon_id(joycon_id), (
		"No accessible JoyCon found via evdev backend. Check connection and /dev/hidraw permissions"
	)

	joycon = JoyCon(*joycon_id)
	status = joycon.get_status()

	assert isinstance(status, dict)
	assert "battery" in status
	assert "buttons" in status
	assert "gyro" in status

	vendor_id, product_id, _serial = joycon_id
	assert vendor_id == JOYCON_VENDOR_ID
	assert product_id in (JOYCON_L_PRODUCT_ID, JOYCON_R_PRODUCT_ID)


def test_evdev_backend_can_read_imu_data():
	"""
	Integration test: validate accelerometer and gyroscope access via evdev path.
	"""
	joycon_id = _get_any_joycon_id()
	assert _is_valid_joycon_id(joycon_id), (
		"No accessible JoyCon found via evdev backend. Check connection and /dev/hidraw permissions"
	)

	joycon = JoyCon(*joycon_id)
	# Wait briefly for the background polling thread to populate fresh report data.
	time.sleep(0.03)

	accel_samples = [
		(
			joycon.get_accel_x(i),
			joycon.get_accel_y(i),
			joycon.get_accel_z(i),
		)
		for i in range(3)
	]
	gyro_samples = [
		(
			joycon.get_gyro_x(i),
			joycon.get_gyro_y(i),
			joycon.get_gyro_z(i),
		)
		for i in range(3)
	]

	for sample in accel_samples:
		_assert_finite_triplet(sample)
	for sample in gyro_samples:
		_assert_finite_triplet(sample)

	# A resting JoyCon should still report gravity on at least one accel axis.
	assert any(abs(v) > 1e-6 for sample in accel_samples for v in sample)


def _avg_triplets(triplets):
	n = len(triplets)
	return (
		sum(t[0] for t in triplets) / n,
		sum(t[1] for t in triplets) / n,
		sum(t[2] for t in triplets) / n,
	)


def print_processed_imu_data(frames=40, interval=0.05):
	"""
	Print processed IMU data (accelerometer in g, gyroscope in deg/s).
	"""
	rows = _print_joycon_permission_diagnostics()
	joycon_id = _get_any_joycon_id()
	if not _is_valid_joycon_id(joycon_id):
		if rows and not any(r["rw"] for r in rows):
			print("[SKIP] JoyCon found but hidraw permission is denied")
			return
		print(
			"[SKIP] No accessible JoyCon found via evdev backend. "
			"Check connection and /dev/hidraw permissions"
		)
		return

	joycon = JoyCon(*joycon_id)
	print("[IMU ] Printing processed data. Press Ctrl+C to stop.")

	try:
		for i in range(frames):
			# 3 IMU sub-samples exist in each report; average them for stable display.
			accel_raw = [
				(
					joycon.get_accel_x(s),
					joycon.get_accel_y(s),
					joycon.get_accel_z(s),
				)
				for s in range(3)
			]
			gyro_raw = [
				(
					joycon.get_gyro_x(s),
					joycon.get_gyro_y(s),
					joycon.get_gyro_z(s),
				)
				for s in range(3)
			]

			ax, ay, az = _avg_triplets(accel_raw)
			gx, gy, gz = _avg_triplets(gyro_raw)

			# JoyCon conversion factors from existing wrappers.
			ax_g, ay_g, az_g = (ax * (4.0 / 0x4000), ay * (4.0 / 0x4000), az * (4.0 / 0x4000))
			gx_dps, gy_dps, gz_dps = (gx * 0.06103, gy * 0.06103, gz * 0.06103)

			print(
				f"[{i + 1:03d}] "
				f"accel(g)=({ax_g:+.4f}, {ay_g:+.4f}, {az_g:+.4f}) "
				f"gyro(deg/s)=({gx_dps:+.2f}, {gy_dps:+.2f}, {gz_dps:+.2f})"
			)

			time.sleep(interval)
	except KeyboardInterrupt:
		print("\n[IMU ] Stopped by user")


def _vec_to_tuple(v):
	if v is None:
		return None
	for attrs in (("x", "y", "z"), ("x", "y")):
		if all(hasattr(v, a) for a in attrs):
			return tuple(float(getattr(v, a)) for a in attrs)
	try:
		seq = tuple(v)
		return tuple(float(x) for x in seq)
	except Exception:
		return None


def _fmt_vec(v):
	t = _vec_to_tuple(v)
	if t is None:
		return "None"
	return "(" + ", ".join(f"{x:+.4f}" for x in t) + ")"


def _close_joycon(joycon):
	if joycon is None:
		return
	close_fn = getattr(joycon, "_close", None)
	if callable(close_fn):
		try:
			close_fn()
		except Exception:
			pass


def _run_status_demo(joycon, frames, interval):
	print("[STEP] README Usage / get_status()")
	for i in range(frames):
		status = joycon.get_status()
		battery = status.get("battery", {})
		buttons = status.get("buttons", {})
		gyro = status.get("gyro", {})
		print(
			f"  [status {i + 1:03d}] "
			f"battery(level={battery.get('level')}, charging={battery.get('charging')}) "
			f"gyro=({gyro.get('x')}, {gyro.get('y')}, {gyro.get('z')}) "
			f"buttons_keys={list(buttons.keys())}"
		)
		time.sleep(interval)


def _run_gyro_demo(joycon, frames, interval):
	print("[STEP] README Gyroscope / GyroTrackingJoyCon")
	for i in range(frames):
		print(
			f"  [gyro {i + 1:03d}] "
			f"pointer={_fmt_vec(joycon.pointer)} "
			f"rotation={_fmt_vec(joycon.rotation)} "
			f"direction={_fmt_vec(joycon.direction)}"
		)
		time.sleep(interval)


def _run_button_event_demo(joycon, frames, interval):
	print("[STEP] README Button events / ButtonEventJoyCon")
	for i in range(frames):
		events = list(joycon.events())
		if events:
			print(f"  [event {i + 1:03d}] events={events}")
		else:
			print(f"  [event {i + 1:03d}] events=[] (press buttons to generate events)")
		time.sleep(interval)


def _run_combined_demo(joycon, frames, interval):
	print("[STEP] README Combining helper classes / MyJoyCon")
	for i in range(frames):
		events = list(joycon.events())
		print(
			f"  [combo {i + 1:03d}] "
			f"pointer={_fmt_vec(joycon.pointer)} "
			f"rotation={_fmt_vec(joycon.rotation)} "
			f"events={events}"
		)
		time.sleep(interval)


def run_readme_feature_loop(
	status_frames=5,
	gyro_frames=10,
	event_frames=15,
	combined_frames=10,
	interval=0.08,
	cycle_pause=0.4,
	cycles=0,
):
	"""
	Run README-described features sequentially and print continuously.

	cycles=0 means infinite loop until Ctrl+C.
	"""
	cycle_index = 0
	print("[LOOP] README feature loop started. Press Ctrl+C to stop.")
	try:
		while cycles <= 0 or cycle_index < cycles:
			cycle_index += 1
			print(f"\n[CYCLE] {cycle_index}")
			rows = _print_joycon_permission_diagnostics()
			joycon_id = _get_any_joycon_id()
			if not _is_valid_joycon_id(joycon_id):
				if rows and not any(r["rw"] for r in rows):
					print("[SKIP] JoyCon detected but hidraw permission denied")
				else:
					print("[SKIP] No accessible JoyCon found")
				time.sleep(max(0.2, cycle_pause))
				continue

			print(f"[INFO] Using JoyCon id={joycon_id}")
			try:
				joycon = ReadmeCombinedJoyCon(*joycon_id, track_sticks=True)
				try:
					_run_status_demo(joycon, status_frames, interval)
					_run_gyro_demo(joycon, gyro_frames, interval)
					_run_button_event_demo(joycon, event_frames, interval)
					_run_combined_demo(joycon, combined_frames, interval)
				finally:
					_close_joycon(joycon)
			except Exception as e:
				if _is_permission_error(e):
					print("[SKIP] Permission denied while running README checks")
				else:
					print(f"[FAIL] README feature loop failed: {e}")
					traceback.print_exc()

			time.sleep(max(0.2, cycle_pause))
	except KeyboardInterrupt:
		print("\n[LOOP] README feature loop stopped by user")


def _run_direct():
	rows = _print_joycon_permission_diagnostics()

	tests = [
		test_evdev_backend_can_read_status,
		test_evdev_backend_can_read_imu_data,
	]

	failures = []
	skips = []
	for test_fn in tests:
		name = test_fn.__name__
		print(f"[RUN ] {name}")
		try:
			test_fn()
		except AssertionError as e:
			msg = str(e)
			if msg.startswith("No accessible JoyCon found via evdev backend"):
				if rows and not any(r["rw"] for r in rows):
					skips.append((name, e))
					print(f"[SKIP] {name}: JoyCon exists but hidraw permission is denied")
					continue
			failures.append((name, e))
			print(f"[FAIL] {name}: {e}")
			traceback.print_exc()
		except Exception as e:
			if _is_permission_error(e):
				skips.append((name, e))
				print(f"[SKIP] {name}: permission denied for hidraw device")
				continue
			failures.append((name, e))
			print(f"[FAIL] {name}: {e}")
			traceback.print_exc()
		else:
			print(f"[ OK ] {name}")

	if failures:
		raise SystemExit(1)

	if skips:
		print(f"Skipped {len(skips)} test(s) due to hidraw permission limits.")

	print("All JoyCon integration tests passed.")


def _drain_button_events(joycon):
	for _ in joycon.events():
		pass


def _wait_expected_button(joycon, expected_event, timeout):
	deadline = time.time() + timeout
	while time.time() < deadline:
		for event_name, state in joycon.events():
			if state != 1:
				continue
			return event_name == expected_event, event_name
		time.sleep(0.01)
	return False, None


def _interactive_button_map_for_side(side, timeout):
	joycon_id = _get_side_joycon_id(side)
	if not _is_valid_joycon_id(joycon_id):
		print(f"[SKIP] {side} JoyCon not found")
		return

	prompts = RIGHT_BUTTON_PROMPTS if side == "R" else LEFT_BUTTON_PROMPTS
	joycon = None
	results = []
	try:
		joycon = ButtonEventJoyCon(*joycon_id, track_sticks=True)
		print(f"\n[BTN ] Start {side} side mapping check")
		print("[BTN ] Press the prompted button directly. Auto-advance on first press.")

		for label, expected in prompts:
			_drain_button_events(joycon)
			print(f"[WAIT] Please press [{side}:{label}] (timeout {timeout:.1f}s) ...")
			ok, observed = _wait_expected_button(joycon, expected, timeout)
			if ok:
				print(f"[ OK ] [{side}:{label}] -> {observed}")
				results.append((label, expected, "PASS", observed))
			elif observed is None:
				print(f"[FAIL] [{side}:{label}] -> timeout")
				results.append((label, expected, "TIMEOUT", None))
			else:
				print(f"[FAIL] [{side}:{label}] expected={expected} observed={observed}")
				results.append((label, expected, "MISMATCH", observed))
	finally:
		_close_joycon(joycon)

	if results:
		passed = sum(1 for _, _, st, _ in results if st == "PASS")
		failed = sum(1 for _, _, st, _ in results if st in ("TIMEOUT", "MISMATCH"))
		skipped = sum(1 for _, _, st, _ in results if st == "SKIP")
		print(f"[SUM ] {side} side: pass={passed} fail={failed} skip={skipped}")


def _interactive_gyro_for_side(side, frames, interval):
	joycon_id = _get_side_joycon_id(side)
	if not _is_valid_joycon_id(joycon_id):
		print(f"[SKIP] {side} JoyCon not found for gyro check")
		return

	joycon = None
	try:
		joycon = GyroTrackingJoyCon(*joycon_id)
		print(f"\n[GYRO] {side} side check")
		print("[GYRO] Axis-by-axis mode: rotate mainly around the prompted axis")

		axis_steps = [
			("X", 0, "Roll around X axis"),
			("Y", 1, "Pitch around Y axis"),
			("Z", 2, "Yaw around Z axis"),
		]

		for axis_name, axis_idx, hint in axis_steps:
			print(f"[GYRO] {side} axis {axis_name}: {hint}")
			peak_focus = 0.0
			peak_cross = 0.0

			for i in range(frames):
				gx = sum(joycon.get_gyro_x(s) for s in range(3)) / 3.0
				gy = sum(joycon.get_gyro_y(s) for s in range(3)) / 3.0
				gz = sum(joycon.get_gyro_z(s) for s in range(3)) / 3.0

				# JoyCon gyroscope conversion factor to deg/s.
				dps = (gx * 0.06103, gy * 0.06103, gz * 0.06103)
				focus = dps[axis_idx]
				cross = [abs(dps[j]) for j in range(3) if j != axis_idx]

				peak_focus = max(peak_focus, abs(focus))
				peak_cross = max(peak_cross, max(cross))

				print(
					f"  [{side} {axis_name} {i + 1:03d}] "
					f"focus={focus:+7.2f} deg/s "
					f"xyz=({dps[0]:+7.2f}, {dps[1]:+7.2f}, {dps[2]:+7.2f})"
				)
				time.sleep(interval)

			if peak_focus > 0:
				separation = peak_focus / max(1e-6, peak_cross)
				print(
					f"[GYRO] {side} axis {axis_name} summary: "
					f"peak_focus={peak_focus:.2f} deg/s "
					f"peak_cross={peak_cross:.2f} deg/s "
					f"separation={separation:.2f}x"
				)
	finally:
		_close_joycon(joycon)


def run_interactive_mapping(button_timeout=6.0, gyro_frames=80, interval=0.05, start_from="buttons"):
	rows = _print_joycon_permission_diagnostics()
	if rows and not any(r["rw"] for r in rows):
		print("[SKIP] JoyCon detected but hidraw permission denied")
		return

	if start_from == "gyro":
		print("[FLOW] Interactive check: gyro only (R side, then L side)")
	else:
		print("[FLOW] Interactive check: buttons then gyro (R side, then L side)")
	print("[FLOW] If a prompt key does not match your press, mapping is wrong.")

	if start_from != "gyro":
		for side in ("R", "L"):
			_interactive_button_map_for_side(side, timeout=button_timeout)

	for side in ("R", "L"):
		_interactive_gyro_for_side(side, frames=gyro_frames, interval=interval)

	print("[DONE] Interactive mapping flow finished")


if __name__ == "__main__":
	parser = argparse.ArgumentParser(description="JoyCon integration test helper")
	parser.add_argument(
		"--mode",
		choices=["interactive-check", "readme-loop", "tests", "print-imu"],
		default="interactive-check",
		help="execution mode",
	)
	parser.add_argument(
		"--print-imu",
		action="store_true",
		help="print processed IMU data instead of running tests",
	)
	parser.add_argument("--frames", type=int, default=40, help="number of IMU frames to print")
	parser.add_argument(
		"--interval",
		type=float,
		default=0.05,
		help="seconds between IMU prints",
	)
	parser.add_argument("--cycles", type=int, default=0, help="cycles for readme-loop mode (0=infinite)")
	parser.add_argument("--status-frames", type=int, default=5, help="status frames per readme cycle")
	parser.add_argument("--gyro-frames", type=int, default=10, help="gyro frames per readme cycle")
	parser.add_argument("--event-frames", type=int, default=15, help="event frames per readme cycle")
	parser.add_argument("--combined-frames", type=int, default=10, help="combined frames per readme cycle")
	parser.add_argument("--cycle-pause", type=float, default=0.4, help="pause between readme cycles")
	parser.add_argument("--button-timeout", type=float, default=6.0, help="seconds to wait for each prompted button")
	parser.add_argument("--gyro-stream-frames", type=int, default=40, help="frames per axis (X/Y/Z) per side for gyro interactive stream")
	parser.add_argument(
		"--start-from",
		choices=["buttons", "gyro"],
		default="buttons",
		help="for interactive-check mode: start from buttons or jump directly to gyro",
	)
	args = parser.parse_args()

	mode = args.mode
	if args.print_imu:
		mode = "print-imu"

	if mode == "print-imu":
		print_processed_imu_data(frames=max(1, args.frames), interval=max(0.01, args.interval))
	elif mode == "tests":
		_run_direct()
	elif mode == "interactive-check":
		run_interactive_mapping(
			button_timeout=max(0.5, args.button_timeout),
			gyro_frames=max(5, args.gyro_stream_frames),
			interval=max(0.01, args.interval),
			start_from=args.start_from,
		)
	else:
		run_readme_feature_loop(
			status_frames=max(1, args.status_frames),
			gyro_frames=max(1, args.gyro_frames),
			event_frames=max(1, args.event_frames),
			combined_frames=max(1, args.combined_frames),
			interval=max(0.01, args.interval),
			cycle_pause=max(0.05, args.cycle_pause),
			cycles=max(0, args.cycles),
		)