Oculus Touch streaming with pythonosc and openvr

oculus_touch_osc_stream.py

# forked from: https://gist.github.com/awesomebytes/75daab3adb62b331f21ecf3a03b3ab46

import time
import openvr
from pythonosc import osc_message_builder
from pythonosc import osc_bundle_builder
from pythonosc import udp_client


def get_controller_ids(vrsys=None):
    if vrsys is None:
        vrsys = openvr.VRSystem()
    else:
        vrsys = vrsys
    left = None
    right = None
    for i in range(openvr.k_unMaxTrackedDeviceCount):
        device_class = vrsys.getTrackedDeviceClass(i)
        if device_class == openvr.TrackedDeviceClass_Controller:
            role = vrsys.getControllerRoleForTrackedDeviceIndex(i)
            if role == openvr.TrackedControllerRole_RightHand:
                right = i
            if role == openvr.TrackedControllerRole_LeftHand:
                left = i
    return left, right


def from_controller_state_to_dict(pControllerState):
    # docs: https://github.com/ValveSoftware/openvr/wiki/IVRSystem::GetControllerState
    d = {}
    d["unPacketNum"] = pControllerState.unPacketNum
    d["trigger"] = pControllerState.rAxis[1].x
    d["trackpad_x"] = pControllerState.rAxis[0].x
    d["trackpad_y"] = pControllerState.rAxis[0].y
    d["button_upper"] = bool(pControllerState.ulButtonPressed >> 1 & 1)
    d["button_lower"] = bool(pControllerState.ulButtonPressed >> 7 & 1)
    d["trackpad_pressed"] = bool(pControllerState.ulButtonPressed >> 32 & 1)
    d["trackpad_touched"] = bool(pControllerState.ulButtonTouched >> 32 & 1)
    d["button_grip"] = bool(pControllerState.ulButtonPressed >> 2 & 1)
    return d


if __name__ == "__main__":
    max_init_retries = 4
    retries = 0
    print("Initializing OpenVR...")
    while retries < max_init_retries:
        try:
            openvr.init(openvr.VRApplication_Scene)
            break
        except openvr.OpenVRError as e:
            print(
                "Error when initializing OpenVR (try {} / {})".format(retries + 1, max_init_retries)
            )
            print(e)
            retries += 1
            time.sleep(2.0)
    else:
        print("Could not initialize OpenVR, aborting.")
        print("Make sure the system is correctly plugged, you can also try")
        print("to do:")
        print("killall -9 vrcompositor vrmonitor vrdashboard")
        print("Before running this program again.")
        exit(0)

    print("Success!")
    vrsystem = openvr.VRSystem()

    left_id, right_id = None, None
    print("Waiting for controllers...")
    try:
        while left_id is None or right_id is None:
            left_id, right_id = get_controller_ids(vrsystem)
            if left_id and right_id:
                break
            print("Waiting for controllers...")
            time.sleep(1.0)
    except KeyboardInterrupt:
        print("Control+C pressed, shutting down...")
        openvr.shutdown()

    print("Left controller ID: " + str(left_id))
    print("Right controller ID: " + str(right_id))

    reading_rate_hz = 100
    last_unPacketNum_left = 0
    last_unPacketNum_right = 0

    print("Printing controller events!")

    # initialize OSC client
    client = udp_client.SimpleUDPClient("127.0.0.1", 7007)

    vals = [
        "trackpad_x",
        "trackpad_y",
        "trigger",
        "trackpad_pressed",
        "button_grip",
        "button_upper",
        "button_lower",
    ]

    try:
        while True:
            time.sleep(1.0 / reading_rate_hz)

            # Initialize OSC bundle for all tracked controllers
            bundle = osc_bundle_builder.OscBundleBuilder(osc_bundle_builder.IMMEDIATELY)

            result, pControllerState = vrsystem.getControllerState(left_id)
            d = from_controller_state_to_dict(pControllerState)
            msg = osc_message_builder.OscMessageBuilder(address="/vr_osc/left_touch")
            for v in vals:
                msg.add_arg(d[v])
                bundle.add_content(msg.build())

            result, pControllerState = vrsystem.getControllerState(right_id)
            d = from_controller_state_to_dict(pControllerState)
            msg = osc_message_builder.OscMessageBuilder(address="/vr_osc/right_touch")
            for v in vals:
                msg.add_arg(d[v])
                bundle.add_content(msg.build())

            # Send the bundle
            client.send(bundle.build())

    except KeyboardInterrupt:
        print("Control+C pressed, shutting down...")
        openvr.shutdown()