Part 1: Setup Interface

The setup interface is the first component of a visual interface in MyoGestic. It is responsible for:

• Configuring VI-specific parameters (ports, executable paths, etc.).

• Launching and stopping the external VI process.

• Managing bi-directional communication (outgoing predictions, incoming kinematics) via UDP or another protocol.

• Providing custom signals and data buffers for the Online protocol.

Step 1: Create the UI

Design a Qt .ui file with at least a groupBox containing a start/stop button. Convert it with pyside6-uic.

Note

Copy an existing UI file from myogestic/gui/widgets/visual_interfaces/virtual_hand_interface/ui/ as a starting point.

Step 2: Understand the Base Class

Your setup interface must inherit from:

SetupInterfaceTemplate(*args, **kwargs)

Base class for the setup interface of a visual interface.

The base class provides:

• outgoing_message_signal (Signal(QByteArray)) – Send data to the VI.

• incoming_message_signal (Signal(np.ndarray)) – Receive data from the VI.

• _main_window – Reference to the main MyoGestic application.

• _record_protocol, _training_protocol, _online_protocol – Protocol references.

You must implement these abstract methods:

• initialize_ui_logic() – Wire up UI widgets.

• start_interface() – Launch the VI process.

• stop_interface() – Stop the VI process.

• interface_was_killed() – Handle unexpected VI termination.

• close_event() – Clean up on application exit.

• connect_custom_signals() / disconnect_custom_signals() – Connect/disconnect data streaming signals.

• clear_custom_signal_buffers() – Reset internal data buffers.

Optionally override:

• get_custom_save_data() – Return extra data to include when saving recordings (e.g., predicted hand positions).

Step 3: Implement the Setup Interface

Below is a minimal self-contained example, followed by references to the full VHI implementation.

Step 3.1: Minimal Setup Interface

This shows the skeleton of a setup interface. In practice, you would add UDP socket management, process launching, and data streaming.

import numpy as np
from PySide6.QtCore import QByteArray
from PySide6.QtGui import QCloseEvent

from myogestic.gui.widgets.templates.visual_interface import SetupInterfaceTemplate


class MyInterface_SetupInterface(SetupInterfaceTemplate):
    """Minimal setup interface example."""

    def __init__(self, main_window, name: str = "MyInterface"):
        # Import your generated UI class:
        # from myogestic.gui.widgets.visual_interfaces.my_interface.ui import Ui_MySetup
        # super().__init__(main_window, name, ui=Ui_MySetup())
        #
        # For this example we pass None (won't run without a real UI):
        pass  # Replace with super().__init__(...)

    def initialize_ui_logic(self) -> None:
        """Wire up buttons, labels, and other UI elements."""
        pass

    def start_interface(self) -> None:
        """Launch the external VI process and start UDP communication."""
        pass

    def stop_interface(self) -> None:
        """Stop the VI process and close sockets."""
        pass

    def interface_was_killed(self) -> None:
        """Handle unexpected VI termination (e.g., user closed window)."""
        pass

    def close_event(self, event: QCloseEvent) -> None:
        """Clean up on application exit."""
        pass

    def connect_custom_signals(self) -> None:
        """Connect data streaming signals (called when online starts)."""
        pass

    def disconnect_custom_signals(self) -> None:
        """Disconnect data streaming signals (called when online stops)."""
        pass

    def clear_custom_signal_buffers(self) -> None:
        """Reset internal data buffers."""
        pass

    def get_custom_save_data(self) -> dict:
        """Return extra data to include when saving recordings."""
        return {}

Step 3.2: VHI Reference – Class Definition and Helpers

The VHI setup interface launches a Unity executable, manages UDP communication on multiple ports, and streams kinematics at 32 Hz.

Imports

import ast
import platform
import subprocess
import sys
import time
from pathlib import Path

import numpy as np
from PySide6.QtCore import QByteArray, QMetaObject, QProcess, QTimer, Qt, Signal
from PySide6.QtGui import QCloseEvent
from PySide6.QtNetwork import QHostAddress, QUdpSocket
from PySide6.QtWidgets import QCheckBox, QPushButton, QWidget

from myogestic.gui.widgets.logger import LoggerLevel
from myogestic.gui.widgets.templates.visual_interface import SetupInterfaceTemplate
from myogestic.gui.widgets.visual_interfaces.virtual_hand_interface.ui import (
    Ui_SetupVirtualHandInterface
)
from myogestic.utils.constants import MYOGESTIC_UDP_PORT

# Stylesheets
NOT_CONNECTED_STYLESHEET = "background-color: red; border-radius: 5px;"
CONNECTED_STYLESHEET = "background-color: green; border-radius: 5px;"

# Constants
STREAMING_FREQUENCY = 32
TIME_BETWEEN_MESSAGES = 1 / STREAMING_FREQUENCY

SOCKET_IP = "127.0.0.1"
STATUS_REQUEST = "status"
STATUS_RESPONSE = "active"


# Ports
# on this port the VHI listens for incoming messages from MyoGestic
VHI__UDP_PORT = 1236

# on this port the VHI sends the currently displayed predicted hand after having applied linear interpolation
VHI_PREDICTION__UDP_PORT = 1234

Constructor

    def __init__(self, main_window, name="VirtualHandInterface"):
        super().__init__(main_window, name, ui=Ui_SetupVirtualHandInterface())

        self._unity_process = QProcess()
        executable = self._get_unity_executable().resolve()  # Get absolute path
        self._unity_process.setProgram(str(executable))

        # On macOS, set working directory to the app's MacOS folder for Unity to find resources
        if platform.system() == "Darwin":
            self._unity_process.setWorkingDirectory(str(executable.parent))

        self._unity_process.started.connect(
            lambda: self._main_window.toggle_selected_visual_interface(self.name)
        )
        self._unity_process.finished.connect(self.interface_was_killed)
        self._unity_process.finished.connect(
            lambda: self._main_window.toggle_selected_visual_interface(self.name)
        )
        self._unity_process.errorOccurred.connect(self._on_process_error)
        self._unity_process.readyReadStandardError.connect(self._on_stderr)

        self._setup_timers()

        self._last_message_time = time.time()
        self._is_connected: bool = False
        self._streaming__udp_socket: QUdpSocket | None = None

        # Custom Stuff
        self._predicted_hand__udp_socket: QUdpSocket | None = None
        self._predicted_hand_recording__buffer: list[(float, np.ndarray)] = []

        # Initialize Virtual Hand Interface UI
        self.initialize_ui_logic()

Step 3.3: VHI Reference – Lifecycle Methods

start_interface – Launch Unity and open UDP sockets

    def start_interface(self):
        """Start the Virtual Hand Interface."""
        if not self._use_external_virtual_hand_interface__check_box.isChecked():
            # Prepare macOS executable (chmod +x and remove quarantine)
            executable = self._get_unity_executable().resolve()
            self._prepare_macos_executable(executable)
            self._unity_process.start()
            self._unity_process.waitForStarted()
        self._status_request__timer.start()
        self.toggle_streaming()

stop_interface – Stop Unity and close sockets

    def stop_interface(self):
        """Stop the Virtual Hand Interface."""
        if not self._use_external_virtual_hand_interface__check_box.isChecked():
            self._unity_process.kill()
            self._unity_process.waitForFinished()
        # In case the stop function would be called from outside the main thread we need to use invokeMethod
        QMetaObject.invokeMethod(self._status_request__timer, "stop", Qt.QueuedConnection)
        self.toggle_streaming()

Step 3.4: VHI Reference – Communication and Streaming

toggle_streaming – Start/stop data streaming

    def toggle_streaming(self) -> None:
        """Toggle the streaming of the Virtual Hand Interface."""
        if self._toggle_virtual_hand_interface__push_button.isChecked():
            self._streaming__udp_socket = QUdpSocket(self)
            self._streaming__udp_socket.readyRead.connect(self.read_message)
            self.outgoing_message_signal.connect(self.write_message)
            self._streaming__udp_socket.bind(
                QHostAddress(SOCKET_IP), MYOGESTIC_UDP_PORT
            )

            self._predicted_hand__udp_socket = QUdpSocket(self)
            self._predicted_hand__udp_socket.bind(
                QHostAddress(SOCKET_IP), VHI_PREDICTION__UDP_PORT
            )
            self._predicted_hand__udp_socket.readyRead.connect(self.read_predicted_hand)

            self._last_message_time = time.time()
        else:
            try:
                self._streaming__udp_socket.close()
                self._predicted_hand__udp_socket.close()
            except AttributeError:
                pass
            self._streaming__udp_socket = None
            self._predicted_hand__udp_socket = None
            self._is_connected = False
            self._virtual_hand_interface__status_widget.setStyleSheet(
                NOT_CONNECTED_STYLESHEET
            )

read_message – Read UDP messages from the VI

    def read_message(self) -> None:
        """Read a message from the Virtual Hand Interface."""
        if self._toggle_virtual_hand_interface__push_button.isChecked():
            while self._streaming__udp_socket.hasPendingDatagrams():
                datagram, _, _ = self._streaming__udp_socket.readDatagram(
                    self._streaming__udp_socket.pendingDatagramSize()
                )

                data = datagram.data().decode("utf-8")
                if not data:
                    return

                try:
                    if data == STATUS_RESPONSE:
                        self._is_connected = True
                        self._virtual_hand_interface__status_widget.setStyleSheet(
                            CONNECTED_STYLESHEET
                        )
                        self._status_request_timeout__timer.stop()
                        return

                    self.incoming_message_signal.emit(np.array(ast.literal_eval(data)))
                except (UnicodeDecodeError, SyntaxError):
                    pass

Step 3.5: VHI Reference – Custom Signals and Data

connect_custom_signals

    def connect_custom_signals(self) -> None:
        """Connect custom signals for the Virtual Hand Interface."""
        self.predicted_hand__signal.connect(self.online_predicted_hand_update)

get_custom_save_data – Extra recording data (predicted hand positions)

    def get_custom_save_data(self) -> dict:
        """Get custom save data for the Virtual Hand Interface."""
        return {
            "predicted_hand": np.vstack(
                [data for _, data in self._predicted_hand_recording__buffer],
            ).T,
            "predicted_hand_timings": np.array(
                [time for time, _ in self._predicted_hand_recording__buffer],
            ),
        }

Step 3.6: Registration in CONFIG_REGISTRY

Both the setup and recording interfaces are registered together. This is typically done in default_config or user_config.

from myogestic.utils.config import CONFIG_REGISTRY
from myogestic.gui.widgets.visual_interfaces.virtual_hand_interface import (
    VirtualHandInterface_SetupInterface,
    VirtualHandInterface_RecordingInterface,
)

CONFIG_REGISTRY.register_visual_interface(
    name="VHI",
    setup_interface_ui=VirtualHandInterface_SetupInterface,
    recording_interface_ui=VirtualHandInterface_RecordingInterface,
)