biosiglive.interfaces package

Submodules

biosiglive.interfaces.generic_interface module

This file contains a generic interface class to use for any new implemented class.

class biosiglive.interfaces.generic_interface.GenericInterface(ip: str = '127.0.0.1', system_rate: float = 100, interface_type: InterfaceType | str = None)[source]

Bases: object

Class for generic interfacing.

add_device(**kwargs)[source]

Add a device to the interface.

add_marker_set(**kwargs)[source]

Add a marker set to stream from the interface system.

get_device(name: str = None, idx: int = None)[source]

Get a device from the interface.

Parameters:
  • idx (int) – Index of the device.

  • name (str) – Name of the device.

Return type:

The device object.

get_device_data(**kwargs)[source]

Get the device data.

static get_force_plate_data()[source]

Get the force plate data.

get_frame()[source]

Get the frame of the interface. That need to be call to retrieve all the data once at the same time.

get_frame_number()[source]

Get the frame number of the interface.

get_kinematics_from_marker_set(**kwargs)[source]

Get the kinematics from a marker set.

get_latency()[source]

Get the latency of the interface.

get_marker_set(name: str = None, idx: int = None)[source]

Get a device from the interface.

Parameters:
  • name (str) – Name of the device.

  • idx (int) – Index of the device.

Return type:

The device.

get_marker_set_data(**kwargs)[source]

Get the marker set data.

init_client()[source]

Initialize the client.

biosiglive.interfaces.param module

This file contains the Parameter class that define the device and markers classes.

class biosiglive.interfaces.param.Device(device_type: DeviceType = DeviceType.Emg, nb_channels: int = 1, name: str = None, rate: float = 2000, system_rate: float = 100, channel_names: list | str = None)[source]

Bases: Param

This class is used to store the available devices.

process(method: str | RealTimeProcessingMethod | OfflineProcessingMethod = None, custom_function: callable = None, **kwargs)[source]

Process the data of the device. The raw data are stored in a buffer fill by the append data method. This method should be called after any get_data method, otherwise there will be not available data.

Parameters:
  • method (GenericProcessing) – Method to process the data.

  • custom_function (callable) – Custom function to process the data.

  • kwargs – Keyword arguments to pass to the method.

class biosiglive.interfaces.param.MarkerSet(nb_channels: int = 1, name: str = None, marker_names: str | list = None, rate: float = None, unlabeled: bool = False, system_rate: float = 100, unit: str = 'm')[source]

Bases: Param

This class is used to store the available markers.

get_kinematics(model_path: str = None, method: InverseKinematicsMethods | str = None, custom_function: callable = None, kin_data_window: int = None, **kwargs) tuple[source]

Function to apply the Kalman filter to the markers. The raw data are stored in a buffer fill by the append data method. This method should be called after any get_data method, otherwise there will be not available data.

Parameters:
  • model_path (str) – The biomod model used to compute the kinematics.

  • method (Union[InverseKinematicsMethods, str]) – The method to use to compute the inverse kinematics.

  • custom_function (callable) – Custom function to use.

  • kin_data_window (int) – The size of the window to use to compute the kinematics.

  • **kwargs (dict) – Keyword arguments to pass to the method.

Returns:

The joint angle and velocity.

Return type:

tuple

class biosiglive.interfaces.param.Param(nb_channels: int, name: str = None, rate: float = None, system_rate: float = 100, data_window: int = None)[source]

Bases: object

append_data(new_data: ndarray)[source]

Append new data to the parameter to the raw data

Parameters:

new_data (np.ndarray) – new data to append to the buffer

biosiglive.interfaces.pytrigno_interface module

class biosiglive.interfaces.pytrigno_interface.PytrignoClient(system_rate=100, ip: str = '127.0.0.1', init_now: bool = True)[source]

Bases: GenericInterface

Class to wrap the Trigno community SDK.

add_device(nb_channels: int, device_type: DeviceType | str = DeviceType.Emg, data_buffer_size: int = None, name: str = None, rate: float = 2000, device_range: tuple = None, processing_method: RealTimeProcessingMethod | OfflineProcessingMethod = None, **process_kwargs)[source]

Add a device to the Pytrigno system.

Parameters:
  • nb_channels (int) – Number of channels of the device.

  • device_type (Union[DeviceType, str]) – Type of the device.

  • data_buffer_size (int) – Size of the buffer for the device.

  • name (str) – Name of the device.

  • rate (float) – Rate of the device.

  • device_range (tuple) – Range of the device. Number of selected channels. If None, all channels are selected (0, 16).

  • processing_method (Union[RealTimeProcessingMethod, OfflineProcessingMethod]) – Method used to process the data.

  • **process_kwargs – Keyword arguments for the processing method.

get_device_data(device_name: str = 'all', channel_idx: int | list = (), get_frame: bool = True) ndarray[source]

Get data from the device.

Parameters:
  • device_name (str) – Name of the device.

  • channel_idx (Union[int, list]) – Index of the channel.

  • get_frame (bool) – Get data from device. If False, use the last data acquired.

Returns:

data – Data from the device.

Return type:

list

get_frame()[source]

Get a frame from the interface. This function is used to get data from the interface.

init_client()[source]

Initialize the client if it’s not already done. This function has to be called before getting a frame.

biosiglive.interfaces.tcp_interface module

This file contains a wrapper to use a tcp client more easily.

class biosiglive.interfaces.tcp_interface.TcpClient(ip: str = '127.0.0.1', port: int = 801, client_type: str = 'TCP', read_frequency: int = 100)[source]

Bases: GenericInterface

Class for interfacing with the client.100

add_device(nb_channels: int, device_type: DeviceType | str = DeviceType.Emg, command_name: str | list = '', name: str = None, rate: float = 2000, device_range: tuple = None, processing_method: RealTimeProcessingMethod | OfflineProcessingMethod = None, **process_kwargs)[source]

Add a device to the client.

Parameters:
  • nb_channels (int) – Number of channels of the device.

  • device_type (Union[DeviceType, str]) – Type of the device. (emg, imu, etc.)

  • command_name (Union[str, list]) – Name of the command to send to the server.

  • name (str) – Name of the device.

  • rate (float) – Frequency of the device.

  • device_range (tuple) – Range of the device.

  • processing_method (Union[RealTimeProcessingMethod, OfflineProcessingMethod]) – Method to use to process the data.

  • process_kwargs (dict) – Keyword arguments for the processing method.

add_marker_set(nb_markers: int, name: str = None, command_name: str | list = None, marker_names: str | list = None, rate: float = 100, unlabeled: bool = False, subject_name: str = None, kinematics_method: InverseKinematicsMethods = None, **kin_method_kwargs)[source]

Add markers set to stream from the tcp interface.

Parameters:
  • nb_markers (int) – Number of markers.

  • name (str) – Name of the markers set.

  • command_name (Union[str, list]) – Name of the command to send to the server.

  • marker_names (Union[list, str]) – List of markers names.

  • rate (int) – Rate of the markers set.

  • unlabeled (bool) – Whether the markers set is unlabeled.

  • subject_name (str) – Name of the subject. If None, the subject will be the first one in Nexus.

  • kinematics_method (InverseKinematicsMethods) – Method used to compute the kinematics.

  • **kin_method_kwargs – Keyword arguments for the kinematics method.

get_data_from_server(command: str | list = 'all', nb_frame_to_get: int = None, down_sampling: dict = None) dict[source]

Get the data from the server.

Parameters:
  • command (Union[str, list]) – Command to send to the server.

  • nb_frame_to_get (int) – Number of frame to get.

  • down_sampling (dict) – Down sampling parameters. Keys are the names of the devices and values are the down sampling rate.

Returns:

data – Data from the server.

Return type:

dict

get_device_data(device_name: list | str = 'all', nb_frame_to_get: int = None, down_sampling: dict = None, get_from_server: bool = True)[source]

Get the data from a device.

Parameters:
  • device_name (Union[list, str]) – Name of the device. All for all the devices.

  • nb_frame_to_get (int) – Number of frame to get.

  • down_sampling (dict) – Down sampling parameters. Keys are the names of the devices and values are the down sampling rate.

  • get_from_server (bool) – Whether to get the data from the server or not.

Returns:

data – The data asked from the server.

Return type:

list

get_marker_set_data(marker_set_name: str | list = 'all', nb_frame_to_get: int = None, down_sampling: dict = None, get_from_server: bool = True)[source]

Get the data from the markers.

Parameters:
  • marker_set_name (Union[list, str]) – Name of the markers set. all for all the markers sets.

  • nb_frame_to_get (int) – Number of frame to get.

  • down_sampling (dict) – Down sampling parameters. Keys are the names of the devices and values are the down sampling rate.

  • get_from_server (bool) – Whether to get the data from the server or not.

Returns:

data – The data asked from the server.

Return type:

list

biosiglive.interfaces.vicon_interface module

This file contains a wrapper for the python Vicon SDK.

class biosiglive.interfaces.vicon_interface.ViconClient(system_rate: int, ip: str = '127.0.0.1', port: int = 801, init_now=True)[source]

Bases: GenericInterface

Class for interfacing with the Vicon system.

add_device(nb_channels: int, device_type: DeviceType | str = DeviceType.Emg, data_buffer_size: int = None, name: str = None, rate: float = 2000, device_range: tuple = None, processing_method: RealTimeProcessingMethod | OfflineProcessingMethod = None, **process_kwargs)[source]

Add a device to the Vicon system.

Parameters:
  • nb_channels (int) – Number of channels of the device.

  • device_type (Union[DeviceType, str]) – Type of the device.

  • data_buffer_size (int) – Size of the buffer for the device.

  • name (str) – Name of the device.

  • rate (float) – Rate of the device.

  • device_range (tuple) – Range of the device.

  • processing_method (Union[RealTimeProcessingMethod, OfflineProcessingMethod]) – Method used to process the data.

  • **process_kwargs – Keyword arguments for the processing method.

add_marker_set(nb_markers: int, name: str = None, data_buffer_size: int = None, marker_names: str | list = None, rate: float = 100, unlabeled: bool = False, subject_name: str = None, kinematics_method: InverseKinematicsMethods = None, **kin_method_kwargs)[source]

Add markers set to stream from the Vicon system.

Parameters:
  • nb_markers (int) – Number of markers.

  • name (str) – Name of the markers set.

  • data_buffer_size (int) – Size of the buffer for the markers set.

  • marker_names (Union[list, str]) – List of markers names.

  • rate (int) – Rate of the markers set.

  • unlabeled (bool) – Whether the markers set is unlabeled.

  • subject_name (str) – Name of the subject. If None, the subject will be the first one in Nexus.

  • kinematics_method (InverseKinematicsMethods) – Method used to compute the kinematics.

  • **kin_method_kwargs – Keyword arguments for the kinematics method.

get_device_data(device_name: str | list = 'all', channel_idx: int | list = (), get_frame: bool = True)[source]

Get the device data from Vicon.

Parameters:
  • device_name (str or list) – Name of the device or list of devices names.

  • channel_idx (Union[int, str]) – Index of the channel to return.

  • get_frame (bool) – Whether to get a new frame from the Vicon system.

Returns:

device_data – All asked device data.

Return type:

list

static get_force_plate_data()[source]

Get the force plate data.

get_frame() bool[source]

Get a new frame from the Vicon system.

Returns:

True if there is a frame, False otherwise.

Return type:

bool

get_frame_number() int[source]

Get the last frame number.

Returns:

frame_number – Last frame number.

Return type:

int

get_kinematics_from_markers(marker_set_name: str, model_path: str = None, method: InverseKinematicsMethods | str = InverseKinematicsMethods.BiorbdLeastSquare, custom_func: callable = None, **kwargs)[source]

Get the kinematics from markers.

Parameters:
  • marker_set_name (str) – name of the markerset.

  • model_path (str) – biorbd model of the kinematics.

  • method (str) – Method to use to get the kinematics. Can be “kalman” or “custom”.

  • custom_func (function) – Custom function to get the kinematics.

Returns:

kinematics – List of kinematics.

Return type:

list

get_latency() float[source]

Get the latency between the Vicon system and the Vicon SDK.

Returns:

latency – Latency between the Vicon system and the Vicon SDK.

Return type:

float

get_marker_set_data(subject_name: str | list = None, marker_names: str | list = None, get_frame: bool = True)[source]

Get the markers data from Vicon.

Parameters:
  • subject_name (Union[str, list]) – Name of the subject. If None, the subject will be the first one in Nexus.

  • marker_names (Union[str, list]) – List of markers names.

  • get_frame (bool) – Whether to get a new frame or not.

Returns:

markers_data – All asked markers data.

Return type:

list

init_client()[source]

Initialize the Vicon client if it is not already initialized. This function has to be called before get frame from interface.

Module contents