Skip to content

tools

Description

This module contains the functions that don't properly apply to the plot or analysis category but that are necessary for the usability of the library. The functions contained in this module can be considered as "tools" or shortcuts necessary to operate with the HD-EMG recordings.


standardise_emgfile_dtypes(emgfile)

Standardise the data types of fields in an emgfile dictionary.

This function ensures that each standard key conforms to a predefined data type specification. It enforces numeric data types, and standardises array and scalar types to maintain consistency across saving/loading and processing of the emgfile.

For motor unit pulse trains, all arrays are verified to be one-dimensional. If a non-1D array is detected, the function issues a warning and automatically flattens it to 1D to preserve compatibility.

The following keys are checked and standardised:

  • "SOURCE" (str)
  • "FILENAME" (str)
  • "RAW_SIGNAL" (pandas.DataFrame, np.float64)
  • "REF_SIGNAL" (pandas.DataFrame, np.float64)
  • "ACCURACY" (pandas.DataFrame, np.float64)
  • "IPTS" (pandas.DataFrame, np.float64)
  • "MUPULSES" (list of 1D numpy.ndarray, np.int64)
  • "FSAMP" (float)
  • "IED" (float)
  • "EMG_LENGTH" (int)
  • "NUMBER_OF_MUS" (int)
  • "BINARY_MUS_FIRING" (pandas.DataFrame, np.uint8)

  • "GOOD_CHANNELS" (dict{str: int})

  • "REFERENCE_MUPULSES" (list of 1D numpy.ndarray, np.int64)
  • "ROA_WITH_REFERENCE_MUPULSES" (pandas.DataFrame, np.float64)

Any additional keys (e.g., "EXTRAS") are preserved but not type-checked.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

RETURNS DESCRIPTION
dict

A deep copy of the input emgfile where all recognised fields have been standardised to the expected data types.

RAISES DESCRIPTION
TypeError

If a recognised field does not match the expected data type or cannot be cast.

WARNS DESCRIPTION
UserWarning

If "MUPULSES"[n] or "REFERENCE_MUPULSES"[n] is not 1D and is flattened.

Examples:

Check data types for IPTS of the sample emgfile.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> print(emgfile["IPTS"].dtypes)
0    float32
1    float32
2    float32
3    float32
4    float32
dtype: object

Standardise the emgfile and check again the data types for IPTS.

>>> standard_emgfile = emg.standardise_emgfile_dtypes(emgfile)
>>> print(standard_emgfile["IPTS"].dtypes)
0    float64
1    float64
2    float64
3    float64
4    float64
dtype: object


showselect(emgfile, how='ref_signal', refsig_channel=0, title='', titlesize=12, nclic=2)

Visually select a part of the recording (X axis).

The area can be selected based on the reference signal or based on the mean EMG signal. Users can move the mouse to track coordinates and press:

- "A" or "a" to add a point at the current cursor location
- "D" or "d" to delete the last selected point
- "Enter" to confirm the selection and close the window

This function does not check whether the selected points are within the effective file duration. This should be done based on user's need.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

how

What to display in the figure used to visually select the area to resize.

ref_signal Visualise the reference signal to select the area to resize.

mean_emg Visualise the mean EMG signal to select the area to resize.

TYPE: str {"ref_signal", "mean_emg"} DEFAULT: "ref_signal"

refsig_channel

The name of the reference signal channel (dataframe column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

title

The title of the plot. It is optional but strongly recommended. It should describe the task to do.

TYPE: str DEFAULT: ''

titlesize

The font size of the title.

TYPE: int DEFAULT: 12

nclic

The number of clics to be collected. If nclic < 1, all the clicks are collected.

DEFAULT: 2

RETURNS DESCRIPTION
points

A list containing the selected points sorted in ascending order.

TYPE: list

Examples:

Load the EMG file and select the points based on the reference signal.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OTB_REFSIG")
>>> points = emg.showselect(
...     emgfile,
...     how="ref_signal",
...     title="Select 2 points",
...     nclic=2,
... )
>>> points
[16115, 40473]

Load the EMG file and select the points based on the mean EMG signal.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OPENHDEMG")
>>> points = emg.showselect(
...     emgfile,
...     how="mean_emg",
...     title="Select 2 points",
...     nclic=2,
... )
>>> points
[135, 26598]


resize_emgfile(emgfile, area=None, how='ref_signal', refsig_channel=0, accuracy='recalculate', compute_on_peaks_only=True, roa_with_reference_mupulses='recalculate', custom_dataframes=None, ignore_negative_ipts=None)

Resize all the STANDARD components in the emgfile.

This function can be useful to compute the various parameters only in the area of interest.

Since version 0.2.0

Specific NON-STANDARD components in the emgfile are also resized.

For STANDARD we refer to:

  • RAW_SIGNAL
  • REF_SIGNAL
  • IPTS
  • MUPULSES
  • EMG_LENGTH
  • BINARY_MUS_FIRING
  • ACCURACY (if recalculated in the new portion)

For NON-STANDARD we refer to:

  • REFERENCE_MUPULSES
  • ROA_WITH_REFERENCE_MUPULSES

Additional dataframes contained in the emgfile will be resized if specified in "custom_dataframes".

Since version 0.2.0

The behaviour of this function has changed to ensure a more accurate SIL estimation. To maintain the old behaviour, you can set compute_on_peaks_only=False when accuracy=="recalculate".

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile to resize.

TYPE: dict

area

The resizing area. If already known, it can be passed in samples, as a list (e.g., [120, 2560]). If None, the user can select the area of interest manually.

TYPE: None or list DEFAULT: None

how

If area==None, allow the user to visually select the area to resize based on how.

ref_signal Visualise the reference signal to select the area to resize.

mean_emg Visualise the mean EMG signal to select the area to resize.

TYPE: str {"ref_signal", "mean_emg"} DEFAULT: "ref_signal"

refsig_channel

The name of the reference signal channel (dataframe column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

accuracy

recalculate The Silhouette score is computed in the new resized file. This can be done only if IPTS is present.

maintain The original accuracy measure already contained in the emgfile is returned without any computation.

TYPE: str {"recalculate", "maintain"} DEFAULT: "recalculate"

compute_on_peaks_only

If True, the silhouette (SIL) score is computed using only the ipts peaks, rather than all values in the source signal. This can improve accuracy estimation by comparing MU spikes only against other candidate spikes, ignoring baseline or negative ipts values. If False, the noise cluster is defined as all samples not selected as MU spikes.

TYPE: bool DEFAULT: True

roa_with_reference_mupulses

Whether to re-calculate the rate of agreement (ROA) between "MUPULSES" and "REFERENCE_MUPULSES".

recalculate The ROA is computed in the new resized file. This can be done only if both "MUPULSES" and "REFERENCE_MUPULSES" are present.

maintain The original ROA measure already contained in the emgfile is returned without any computation.

TYPE: str {"recalculate", "maintain"} DEFAULT: "recalculate"

custom_dataframes

A list of strings pointing to the additional dataframes to resize. The strings should match the emgfile keys associated to the pd.DataFrames.

TYPE: list or None DEFAULT: None

ignore_negative_ipts

This parameter is deprecated and will be removed in future releases. Please transform the 'ipts' before if needed. To replicate the behaviour of 'ignore_negative_ipts=True' you can use 'ipts * np.abs(ipts)'.

TYPE: None DEFAULT: None

RETURNS DESCRIPTION
rs_emgfile

the new (resized) emgfile.

TYPE: dict

start_, end_ : int

the start and end of the selection (can be used for code automation).

Notes

Suggested names for the returned objects: rs_emgfile, start_, end_.

Examples:

Manually select the area to resize the emgfile based on mean EMG signal and recalculate the silhouette score in the new portion of the signal.

>>> emgfile = emg.askloadmodule()
>>> rs_emgfile, start_, end_ = emg.resize_emgfile(
...     emgfile,
...     how="mean_emg",
...     accuracy="recalculate",
... )

Automatically resize the emgfile in the pre-specified area. Do not recalculate the silhouette score in the new portion of the signal.

>>> emgfile = emg.askopenfile(filesource="CUSTOMCSV")
>>> rs_emgfile, start_, end_ = emg.resize_emgfile(
...     emgfile,
...     area=[120, 25680],
...     accuracy="maintain",
... )


select_bad_channels(emgfile, manual_offset=0)

Select noisy channels via visual inspection.

This function opens a modal graphical dialog that allows the user to visually inspect stacked EMG channels and mark noisy or unwanted channels. Channel selection is performed interactively; the calling code is blocked until the dialog is closed.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

manual_offset

This parameter sets the scaling of the channels. If 0 (default), the channels' amplitude is scaled automatically to fit the plotting window. If > 0, the channels will be scaled based on the specified value.

TYPE: int or float DEFAULT: 0

RETURNS DESCRIPTION
edited_emgfile

The EMG file dictionary with an updated "GOOD_CHANNELS" entry (mapping channel indices as strings to booleans) if the user confirms the selection. Returns None if the dialog is cancelled.

TYPE: dict or None

Examples:

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> edited_emgfile = emg.select_bad_channels(emgfile)
>>> if edited_emgfile is not None:
...     print(edited_emgfile["GOOD_CHANNELS"])
>>> else:
...     print("Selection cancelled by the user")


EMGFileSectionsIterator

An iterator for splitting a file into sections and performing actions.

This iterator can be used to split the emgfile (or emg_refsig file) in multiple sections, to apply specific funtions to each of these sections and to gather their results.

This class has a number of methods that help in the splitting process, in the iteration of the various sections, and in merging the results.

PARAMETER DESCRIPTION
file

The dictionary containing the emgfile (or emg_refsig file).

TYPE: dict

ATTRIBUTE DESCRIPTION
file

The dictionary containing the file to split and iterate.

TYPE: dict

file_length

The file duration in samples.

TYPE: int

split_points

A list of sample indices where the file should be split into sections.

TYPE: list of int

sections

A list of sections of the file, created based on split_points.

TYPE: list

results

A list to store the results of operations applied to each section of the file.

TYPE: list

METHOD DESCRIPTION
set_split_points_by_showselect

Manually set the points used to split the emgfile.

set_split_points_by_equal_spacing

Set the points used to split the emgfile into equal sections.

set_split_points_by_time

Set the points used to split the emgfile based on a fixed time window.

set_split_points_by_samples

Set the points used to split the emgfile based on a samples window.

set_split_points_by_list

Set the points used to split the emgfile based on a provided list of sample indices.

split

Splits the file into sections using the set split points.

iterate

Apply a collection of functions to the split sections, each with its own arguments.

merge_dataframes

Merge a list of result DataFrames using the specified method.

set_split_points_by_showselect(how='ref_signal', refsig_channel=0, title='', titlesize=10, nclic=-1)

Manually set the points used to split the emgfile.

Calls the emg.showselect() function to manually select the split points based on the visualisation of the reference signal or of the EMG signal amplitude.

Users can move the mouse to track coordinates and press:

  • "A" or "a" to add a point at the current cursor location
  • "D" or "d" to delete the last selected point
  • "Enter" to confirm the selection and close the window

Sections are cut starting from the first point and then on the consecutive points.

PARAMETER DESCRIPTION
how

What to display in the figure used to visually select the area to resize.

ref_signal Visualise the reference signal to select the area to resize.

mean_emg Visualise the mean EMG signal to select the area to resize.

TYPE: str {"ref_signal", "mean_emg"} DEFAULT: "ref_signal"

refsig_channel

The name of the reference signal channel (column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

title

The title of the plot. It is optional but strongly recommended. It should describe the task to do. A default title is provided when title="".

TYPE: str DEFAULT: ''

titlesize

The font size of the title.

TYPE: int DEFAULT: 12

nclic

The number of clics to be collected. If nclic < 1, all the clicks are collected.

DEFAULT: -1

RETURNS DESCRIPTION
None

Stores the split points in self.split_points.

RAISES DESCRIPTION
ValueError

When the user clicked a wrong number of inputs in the GUI.

Examples:

Manually set the points to resize the file by visualising the reference signal.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_showselect(how="ref_signal")
>>> split_points = iterator.split_points
>>> split_points
[0, 6562, 19552, 41546, 55273, 62802]

Manually set the points to resize the file by visualising the mean EMG signal amplitude.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_showselect(how="mean_emg")
>>> split_points = iterator.split_points
>>> split_points
[5381, 23094, 38889, 50107]

set_split_points_by_equal_spacing(n_sections)

Set the points used to split the emgfile into equal sections.

All the sections will have approximately the same length (length rounding may apply), which is calculated based on n_sections.

PARAMETER DESCRIPTION
n_sections

The number of sections to divide the emgfile into.

TYPE: int

RETURNS DESCRIPTION
None

Stores the split points in self.split_points.

Examples:

Divide the file in 3 sections of the same length.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> split_points = iterator.split_points
>>> split_points
[0, 22186, 44373, 66560]

set_split_points_by_time(time_window, drop_shorter=False)

Set the points used to split the emgfile based on a fixed time window.

PARAMETER DESCRIPTION
time_window

The duration of each section in seconds.

TYPE: float

drop_shorter

If True, the last section is discarded if it is shorter than time_window. If False, the last section will include the remaining samples even if it is shorter than time_window.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
None

Stores the split points in self.split_points.

Examples:

Divide the file into consecutive 9-second sections, with any remaining data forming a final shorter section.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_time(
...     time_window=9,
...     drop_shorter=False,
... )
>>> split_points = iterator.split_points
>>> split_points
[0, 18432, 36864, 55296, 66560]

Divide the file into consecutive 9-second sections, discarding any remaining data if it is shorter than the specified duration.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_time(
...     time_window=9,
...     drop_shorter=True,
... )
>>> split_points = iterator.split_points
>>> split_points
[0, 18432, 36864, 55296]

set_split_points_by_samples(samples_window, drop_shorter=False)

Set the points used to split the emgfile based on a samples window.

PARAMETER DESCRIPTION
samples_window

The duration of each section in samples.

TYPE: int

drop_shorter

If True, the last section is discarded if it is shorter than samples_window. If False, the last section will include the remaining samples even if it is shorter than samples_window.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
None

Stores the split points in self.split_points.

Examples:

Divide the file into consecutive 9-second sections, with any remaining data forming a final shorter section.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_samples(
...     samples_window=10000,
...     drop_shorter=False,
... )
>>> split_points = iterator.split_points
>>> split_points
[0, 10000, 20000, 30000, 40000, 50000, 60000, 66560]

Divide the file into consecutive 9-second sections, discarding any remaining data if it is shorter than the specified duration.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_samples(
...     samples_window=10000,
...     drop_shorter=True,
... )
>>> split_points = iterator.split_points
>>> split_points
[0, 10000, 20000, 30000, 40000, 50000, 60000]

set_split_points_by_list(split_points)

Set the points used to split the emgfile based on a provided list of sample indices.

PARAMETER DESCRIPTION
split_points

A list containing the sample indices at which to split the emgfile. These indices should correspond to the points where the data will be divided into sections.

TYPE: list of int

RETURNS DESCRIPTION
None

Stores the split points in self.split_points.

Examples:

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_list(split_points=[0, 18432, 36864])
>>> split_points = iterator.split_points
>>> split_points
[0, 18432, 36864]

split(accuracy='recalculate', compute_on_peaks_only=True, roa_with_reference_mupulses='recalculate', custom_dataframes=None, ignore_negative_ipts=None)

Splits the file into sections using the set split points.

PARAMETER DESCRIPTION
accuracy

recalculate The Silhouette score is computed in the new resized file. This can be done only if IPTS is present.

maintain The original accuracy measure already contained in the emgfile is returned without any computation.

TYPE: str {"recalculate", "maintain"} DEFAULT: "recalculate"

compute_on_peaks_only

If True, the silhouette (SIL) score is computed using only the ipts peaks, rather than all values in the source signal. This can improve accuracy estimation by comparing MU spikes only against other candidate spikes, ignoring baseline or negative ipts values. If False, the noise cluster is defined as all samples not selected as MU spikes.

TYPE: bool DEFAULT: True

roa_with_reference_mupulses

Whether to re-calculate the rate of agreement (ROA) between "MUPULSES" and "REFERENCE_MUPULSES".

recalculate The ROA is computed in the new resized file. This can be done only if both "MUPULSES" and "REFERENCE_MUPULSES" are present.

maintain The original ROA measure already contained in the emgfile is returned without any computation.

TYPE: str {"recalculate", "maintain"} DEFAULT: "recalculate"

custom_dataframes

A list of strings pointing to the additional dataframes to resize. The strings should match the emgfile keys associated to the pd.DataFrames.

TYPE: list or None DEFAULT: None

ignore_negative_ipts

This parameter is deprecated and will be removed in future releases. Please transform the 'ipts' before if needed. To replicate the behaviour of 'ignore_negative_ipts=True' you can use 'ipts * np.abs(ipts)'.

TYPE: None DEFAULT: None

RETURNS DESCRIPTION
None

Stores the split sections in self.sections.

Examples:

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=4)
>>> iterator.split()
>>> sections = iterator.sections
>>> len(sections)
4

iterate(funcs=[], args_list=[[]], kwargs_list=[{}], **kwargs)

Apply a collection of functions to the split sections, each with its own arguments.

PARAMETER DESCRIPTION
funcs

A list of functions to apply to each section. If multiple functions are provided, their count must match the number of sections. If only one function is given, it will be applied to all sections. IMPORTANT! Each function must take file as the first parameter.

TYPE: list of callables DEFAULT: []

args_list

A list where each element is a list of positional arguments to be passed to the corresponding function in funcs. Must have the same length as funcs.

TYPE: list of lists DEFAULT: [[]]

kwargs_list

A list where each element is a dictionary of keyword arguments to be passed to the corresponding function in funcs. Must have the same length as funcs.

TYPE: list of dicts DEFAULT: [{}]

**kwargs

Additional keyword arguments that are passed to all functions in funcs. If funcs contains only 1 function, **kwargs can be used instead of kwargs_list for simpler syntax.

DEFAULT: {}

RETURNS DESCRIPTION
None

Stores the results in self.results, where each function's output is collected.

Examples:

Split the file in 3 sections and apply a custom function to each section to count the number of firings in each MU. Then visualise the results for the first section.

>>> import openhdemg.library as emg
>>> import pandas as pd
>>> def count_firings(emgfile):
...     res = [len(mu_firings) for mu_firings in emgfile["MUPULSES"]]
...     return pd.DataFrame(res)
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> iterator.split()
>>> iterator.iterate(funcs=[count_firings])
>>> results = iterator.results
>>> results[0]
    0
0  48
1  43
2  63
3  94
4  95

Split the file in 3 sections and calculate the discharge rate of each MU over the first 20 discharges.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> iterator.split()
>>> iterator.iterate(
...     funcs=[emg.compute_dr],
...     event_="rec",
...     n_firings_RecDerec=20,
... )
>>> results = iterator.results
>>> results[0]
     DR_rec     DR_all
0  7.468962   7.714276
1  7.091045   7.390155
2  7.673784   8.583193
3  9.004878  11.042002
4  9.705901  11.202489

merge_dataframes(method='long', fillna=None, agg_func=None)

Merge a list of result DataFrames using the specified method.

PARAMETER DESCRIPTION
method

The merging method. When using built-in methods (except for custom), all DataFrames must have the same structure (i.e., aligned columns and index).

average Computes the mean across all DataFrames.

median Computes the median across all DataFrames.

sum Computes the sum across all DataFrames.

min Takes the minimum value across all DataFrames.

max Takes the maximum value across all DataFrames.

std Computes the standard deviation across all DataFrames.

cv Computes the coefficient of variation (CV = std / mean).

long Stacks all DataFrames with an additional 'source_idx' column.

custom Uses a user-defined aggregation function provided via agg_func.

TYPE: str DEFAULT: "long"

fillna

If specified, fills missing values (NaN) with this value before merging.

TYPE: float or None DEFAULT: None

agg_func

A custom aggregation function to apply when method="custom". The function should take a list of DataFrames and return a single DataFrame.

TYPE: callable or None DEFAULT: None

RETURNS DESCRIPTION
merged_df

The merged DataFrame.

TYPE: DataFrame

RAISES DESCRIPTION
ValueError

If self.results is empty or contains non-DataFrame elements; or, method is unknown; or agg_func is missing when method="custom".

Examples:

Merge all the results in a long format DataFrame. Best for statistical analyses.

>>> import openhdemg.library as emg
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> iterator.split()
>>> iterator.iterate(funcs=[emg.compute_dr], event_="rec")
>>> merged_results = iterator.merge_dataframes()
>>> merged_results
    source_idx  original_idx     DR_rec     DR_all
0            0             0   3.341579   7.714276
1            0             1   5.701081   7.390155
2            0             2   5.699017   8.583193
3            0             3   7.548770  11.042002
4            0             4   8.344515  11.202489
5            1             0  10.235710   8.155868
6            1             1   6.769358   6.758350
7            1             2   8.193645   8.054868
8            1             3  10.952495  11.151536
9            1             4  11.012249  10.691432
10           2             0   6.430406   6.899233
11           2             1   6.714442   6.274404
12           2             2   7.057244   6.881602
13           2             3  10.577538   9.578987
14           2             4   9.708064   9.562182

Apply a custom function to each section to count the number of firings in each MU, then get the mean and STD values across the 3 sections (just for didactical purposes).

>>> import openhdemg.library as emg
>>> import pandas as pd
>>> def count_firings(emgfile):
...     res = [len(mu_firings) for mu_firings in emgfile["MUPULSES"]]
...     return pd.DataFrame(res)
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> iterator.split()
>>> iterator.iterate(funcs=[count_firings])
>>> mean_values = iterator.merge_dataframes(
...     method="average",
...     fillna=0,
... )
>>> std_values = iterator.merge_dataframes(
...     method="std",
...     fillna=0,
... )
>>> mean_values
           0
0  45.666667
1  51.333333
2  65.666667
3  97.666667
4  97.333333
>>> std_values
           0
0   4.932883
1  18.009257
2  20.132892
3  20.744477
4  16.623277

Apply a custom method by providing an external aggregation function which finds the maximum value at each position and the index of the DataFrame containing it.

>>> import openhdemg.library as emg
>>> import pandas as pd
>>> def max_with_source(results_dataframes):
...     stacked = pd.concat(
...         results_dataframes, keys=range(len(results_dataframes))
...     )
...     max_values = stacked.groupby(level=1).max()
...     max_indices = stacked.groupby(level=1).idxmax().iloc[:, 0]
...     max_values["source_idx"] = max_indices.map(lambda x: x[0])
...     return max_values
>>> emgfile = emg.emg_from_samplefile()
>>> iterator = emg.EMGFileSectionsIterator(file=emgfile)
>>> iterator.set_split_points_by_equal_spacing(n_sections=3)
>>> iterator.split()
>>> iterator.iterate(funcs=[emg.compute_dr], event_="rec")
>>> max_values_with_source = iterator.merge_dataframes(
...     method="custom",
...     fillna=0,
...     agg_func=max_with_source,
... )
>>> max_values_with_source
      DR_rec     DR_all  source_idx
0  10.235710   8.155868           1
1   6.769358   7.390155           1
2   8.193645   8.583193           1
3  10.952495  11.151536           1
4  11.012249  11.202489           1


create_binary_firings(emg_length, number_of_mus, mupulses)

Create a binary representation of the MU firing.

Create a binary representation of the MU firing over time based on the times of firing of each MU.

PARAMETER DESCRIPTION
emg_length

Number of samples (length) in the emg file.

TYPE: int

number_of_mus

Number of MUs in the emg file.

TYPE: int

mupulses

Each ndarray should contain the times of firing (in samples) of each MU.

TYPE: list of ndarrays

RETURNS DESCRIPTION
binary_MUs_firing

A pd.DataFrame containing the binary representation of MUs firing. Please note that dtype=np.uint8. Please convert before processing.

TYPE: DataFrame


mupulses_from_binary(binarymusfiring)

Extract the MUPULSES from the binary MUs firings.

PARAMETER DESCRIPTION
binarymusfiring

A pd.DataFrame containing the binary representation of MUs firings.

TYPE: DataFrame

RETURNS DESCRIPTION
MUPULSES

A list of ndarrays containing the firing time (in samples) of each MU.

TYPE: list


compute_idr(emgfile)

Compute the IDR.

This function computes the instantaneous discharge rate (IDR) from the MUPULSES. The IDR is very useful for plotting and visualisation of the MUs behaviour.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

RETURNS DESCRIPTION
idr

A dict containing a pd.DataFrame for each MU (keys are integers). Accessing the key, we have a pd.DataFrame containing:

  • mupulses: firing sample.
  • diff_mupulses: delta between consecutive firing samples.
  • timesec: delta between consecutive firing samples in seconds.
  • idr: instantaneous discharge rate.

TYPE: dict

Examples:

Load the EMG file, compute IDR and access the results for the first MU.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OTB", otb_ext_factor=8)
>>> idr = emg.compute_idr(emgfile=emgfile)
>>> munumber = 0
>>> idr[munumber]
    mupulses  diff_mupulses    timesec       idr
0        9221            NaN   4.502441       NaN
1        9580          359.0   4.677734  5.704735
2        9973          393.0   4.869629  5.211196
3       10304          331.0   5.031250  6.187311
4       10617          313.0   5.184082  6.543131
..        ...            ...        ...       ...
149     54521          395.0  26.621582  5.184810
150     54838          317.0  26.776367  6.460568
151     55417          579.0  27.059082  3.537133
152     55830          413.0  27.260742  4.958838
153     56203          373.0  27.442871  5.490617


delete_mus(emgfile, munumber, if_single_mu='ignore', delete_delsys_muaps=True)

Delete unwanted MUs.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

munumber

The MUs to remove. If a single MU has to be removed, this should be an int (number of the MU). If multiple MUs have to be removed, a list of int should be passed. An unpacked (*) range can also be passed as munumber=[*range(0, 5)]. munumber is expected to be with base 0 (i.e., the first MU in the file is the number 0).

TYPE: int, list of int

if_single_mu

A string indicating how to behave in case of a file with a single MU.

ignore Ignore the process and return the original emgfile. (Default)

remove Remove the MU and return the emgfile without the MU. This should allow full compatibility with the use of this file in following processing (i.e., save/load and analyse).

TYPE: str {"ignore", "remove"} DEFAULT: "ignore"

delete_delsys_muaps

If true, deletes also the associated MUAPs computed by the Delsys software stored in emgfile["EXTRAS"].

TYPE: Bool DEFAULT: True

RETURNS DESCRIPTION
emgfile

The dictionary containing the emgfile without the unwanted MUs.

TYPE: dict

Examples:

Delete MUs 1,4,5 from the emgfile.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OTB", otb_ext_factor=8)
>>> emgfile = emg.delete_mus(emgfile=emgfile, munumber=[1,4,5])


delete_empty_mus(emgfile)

Delete all the MUs without firings.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

RETURNS DESCRIPTION
emgfile

The dictionary containing the emgfile without the empty MUs.

TYPE: dict


find_duplicates_within(emgfile, correlation_max_lag=0.05, peak_window_half_width=0.0025, duplicate_threshold=30, which='accuracy')

Find duplicate MUs within the same file based on discharge times.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

correlation_max_lag

Maximum lag (in seconds) used when computing the cross-correlation between MU spike trains. Defines the full search range for possible synchronisation peaks. Larger values allow detection of synchrony over wider time shifts. This must be < 1.

TYPE: float DEFAULT: 50e-3

peak_window_half_width

Half-width (in seconds) of the window used around the cross-correlation peak to compute the duplication sensitivity metric. This window should capture the narrow temporal jitter expected for duplicate MUs. This must be < 1 and < correlation_max_lag.

TYPE: float DEFAULT: 2.5e-3

duplicate_threshold

Threshold (in percent) for classifying two MUs as duplicates. The sensitivity metric is computed as the sum of the cross-correlation values within a ±peak_window_half_width window around the correlation peak, normalised by the size of the larger spike train.

TYPE: float DEFAULT: 30

which

How to classify the duplicated MUs.

accuracy The MU with the lowest accuracy is considered duplicate. The emgfile must already contain the 'ACCURACY' dataframe.

covisi The MU with the highest CoV of interspike interval is is considered duplicate.

TYPE: str {"accuracy", "covisi"} DEFAULT: "accuracy"

RETURNS DESCRIPTION
duplicates

Sorted list of MU indices classified as duplicates and therefore recommended for removal.

TYPE: list of int

duplicates_info

Detailed information about each detected duplicate MU pair. Each element of the list is a dictionary describing one MU-MU comparison that exceeded the duplication threshold. The dictionary contains:

pair : tuple of int A tuple "(mu1, mu2)" containing the indices of the two MUs identified as potential duplicates.

accuracy : tuple of float, optional Present only when which="accuracy". Contains the SIL (accuracy) values of the two units in the same order as "pair". The MU with the lowest SIL is considered the duplicate.

covisi : tuple of float, optional Present only when which="covisi". Contains the CoV of the interspike interval for the two units, in the same order as "pair". The MU with the highest CoV-ISI is considered the duplicate.

TYPE: list of dict

See also
  • remove_duplicates_within : Remove duplicate MUs within the same file based on discharge times.

Examples:

Starting from a generic file, the results will look similar to the following if which="accuracy":

>>> duplicates, duplicate_info = find_duplicates_within(
...     emgfile,
...     which="accuracy",
... )
>>> duplicates
[0, 3, 4]
>>> duplicate_info
[
    {
        'pair': (0, 1),
        'accuracy': (0.8768360226084906, 0.9552696780856847)
    },
    {
        'pair': (1, 3),
        'accuracy': (0.9552696780856847, 0.8969855881081578)
    },
    {
        'pair': (1, 4),
        'accuracy': (0.9552696780856847, 0.9178590868820631)
    }
]

Or similar to the following if which="covisi":

>>> duplicates, duplicate_info = find_duplicates_within(
...     emgfile,
...     which="covisi",
... )
>>> duplicates
[0, 1, 2, 3]
>>> duplicate_info
[
    {
        'pair': (0, 1),
        'covisi': (76.9574103452168, 16.266055150945718)
    },
    {
        'pair': (1, 3),
        'covisi': (16.266055150945718, 19.07156533711372)
    },
    {
        'pair': (1, 4),
        'covisi': (16.266055150945718, 15.38224044131706)
    },
    {
        'pair': (2, 4),
        'covisi': (23.264925152223373, 15.38224044131706)
    }
]

The returned duplicates list can be used later to remove the duplicate units by passing it to the delete_mus function:

>>> duplicates, duplicate_info = find_duplicates_within(
...     emgfile,
...     which="accuracy",
... )
>>> emgfile = emg.delete_mus(emgfile, munumber=duplicates)


remove_duplicates_within(emgfile, correlation_max_lag=0.05, peak_window_half_width=0.0025, duplicate_threshold=30, which='accuracy')

Remove duplicate MUs within the same file based on discharge times.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

correlation_max_lag

Maximum lag (in seconds) used when computing the cross-correlation between MU spike trains. Defines the full search range for possible synchronisation peaks. Larger values allow detection of synchrony over wider time shifts. This must be < 1.

TYPE: float DEFAULT: 50e-3

peak_window_half_width

Half-width (in seconds) of the window used around the cross-correlation peak to compute the duplication sensitivity metric. This window should capture the narrow temporal jitter expected for duplicate MUs. This must be < 1 and < correlation_max_lag.

TYPE: float DEFAULT: 2.5e-3

duplicate_threshold

Threshold (in percent) for classifying two MUs as duplicates. The sensitivity metric is computed as the sum of the cross-correlation values within a ±peak_window_half_width window around the correlation peak, normalised by the size of the larger spike train.

TYPE: float DEFAULT: 30

which

How to remove the duplicated MUs.

accuracy The MU with the lowest accuracy is removed. The emgfile must already contain the 'ACCURACY' dataframe.

covisi The MU with the highest CoV of interspike interval is removed.

TYPE: str {"accuracy", "covisi"} DEFAULT: "accuracy"

RETURNS DESCRIPTION
emgfile

The dictionary containing the emgfile without duplicated MUs.

TYPE: dict

See also
  • find_duplicates_within : Find duplicate MUs within the same file based on discharge times.
  • remove_duplicates_between : Remove duplicated MUs across two different files based on STA.


sort_mus(emgfile)

Sort the MUs in order of recruitment.

The following emgfile keys are sorted:

  • IPTS
  • MUPULSES
  • BINARY_MUS_FIRING
  • ACCURACY
  • REFERENCE_MUPULSES
  • ROA_WITH_REFERENCE_MUPULSES
  • MU_LABELS
PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

RETURNS DESCRIPTION
sorted_emgfile

The dictionary containing the sorted emgfile.

TYPE: dict


compute_covsteady(emgfile, start_steady=-1, end_steady=-1, refsig_channel=0)

Calculate the CoV of REF_SIGNAL during the steady-state phase.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

start_steady

The start and end point (in samples) of the steady-state phase. If < 0 (default), the user will need to manually select the start and end of the steady-state phase.

TYPE: int DEFAULT: -1

end_steady

The start and end point (in samples) of the steady-state phase. If < 0 (default), the user will need to manually select the start and end of the steady-state phase.

TYPE: int DEFAULT: -1

refsig_channel

The name of the reference signal channel (dataframe column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

RETURNS DESCRIPTION
covsteady

The coefficient of variation of the steady-state phase in %.

TYPE: float

See also
  • compute_idr : computes the instantaneous discharge rate.

Examples:

Load the EMG file, compute covsteady and access the result from GUI.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OTB", otb_ext_factor=8)
>>> covsteady = emg.compute_covsteady(emgfile=emgfile)
>>> covsteady
1.452806

The process can be automated by bypassing the GUI.

>>> import openhdemg.library as emg
>>> emgfile = emg.askopenfile(filesource="OTB", otb_ext_factor=8)
>>> covsteady = emg.compute_covsteady(
...     emgfile=emgfile,
...     start_steady=3580,
...     end_steady=15820,
... )
>>> covsteady
35.611263


filter_rawemg(emgfile, order=2, lowcut=20, highcut=500)

Band-pass filter the RAW_SIGNAL.

The filter is a Zero-lag band-pass Butterworth.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

order

The filter order. Note that a band-pass transformation doubles the order, so order=2 produces a 4th-order band-pass filter.

TYPE: int DEFAULT: 2

lowcut

The lower cut-off frequency in Hz.

TYPE: int DEFAULT: 20

highcut

The higher cut-off frequency in Hz.

TYPE: int DEFAULT: 500

RETURNS DESCRIPTION
filteredrawsig

The dictionary containing the emgfile with a filtered RAW_SIGNAL.

TYPE: dict

See also
  • filter_refsig : low-pass filter the REF_SIGNAL.
Notes

Currently, the returned filteredrawsig cannot be accurately compressed when using the functions save_json_emgfile() and asksavefile(). We therefore suggest you to use high-level functions and classes for binary files such as: 'save_openhdemg_module', 'asksavemodule', 'load_openhdemg_module', 'askopenmodule', 'openhdemg_Collection'.


filter_refsig(emgfile, order=4, cutoff=15, refsig_channels=[0])

Low-pass filter the REF_SIGNAL.

This function is used to low-pass filter the REF_SIGNAL and remove noise. The filter is a Zero-lag low-pass Butterworth applied to the selected channels.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

order

The effective filter order.

TYPE: int DEFAULT: 4

cutoff

The cut-off frequency in Hz.

TYPE: int DEFAULT: 15

refsig_channels

The reference signal channels (DataFrame columns) to filter.

TYPE: list DEFAULT: [0]

RETURNS DESCRIPTION
filteredrefsig

The dictionary containing the emgfile with a filtered REF_SIGNAL.

TYPE: dict

See also
  • remove_offset : remove the offset from the REF_SIGNAL.
  • filter_rawemg : band-pass filter the RAW_SIGNAL.


remove_powerline_harmonics(sig, fsamp, notch_freq=50.0, notch_width=5.0)

Remove power-line interference by zeroing FFT bins around all harmonics of the mains frequency.

PARAMETER DESCRIPTION
sig

2-D array of shape (n_channels, n_samples). Each row is a signal and each column is a time sample.

TYPE: ndarray

fsamp

Sampling frequency in Hz.

TYPE: float

notch_freq

Fundamental power-line frequency (e.g., 50 or 60 Hz).

TYPE: float DEFAULT: 50.0

notch_width

Width of each notch (± notch_width/2), in Hz.

TYPE: float DEFAULT: 5.0

RETURNS DESCRIPTION
ndarray

Filtered signals with the same shape as sig.

Examples:

Remove powerline harmonics in emgfile["RAW_SIGNAL"].

>>> import openhdemg.library as emg
>>> import pandas as pd
>>> import numpy as np
>>> emgfile = emg.emg_from_samplefile()
>>> filtered_sig = emg.remove_powerline_harmonics(
...     sig=np.transpose(emgfile["RAW_SIGNAL"].to_numpy()),
...     fsamp=emgfile["FSAMP"],
... )
>>> emgfile["RAW_SIGNAL"] = pd.DataFrame(
...     filtered_sig.T,
...     dtype=np.float64,
... )


remove_offset(emgfile, offsetval=0, auto=0, refsig_channels=[0])

Remove the offset from the REF_SIGNAL.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

offsetval

Value of the offset(s) to subtract.

  • If a single value (float/int), the same offset is subtracted from all selected channels.
  • If a list, it must have the same length as refsig_channels and each value is subtracted from the corresponding channel.
  • If 0, the user is asked to manually select an area to compute the offset (one channel at a time).

TYPE: float or list DEFAULT: 0

auto

If auto > 0, automatically compute and remove the offset using the first auto samples.

TYPE: int DEFAULT: 0

refsig_channels

The reference signal channels (DataFrame columns) to process.

TYPE: list DEFAULT: [0]

RETURNS DESCRIPTION
offs_emgfile

The dictionary containing the emgfile with a corrected offset of the REF_SIGNAL.

TYPE: dict

See also
  • filter_refsig : low-pass filter REF_SIGNAL.


get_mvc(emgfile, how='showselect', conversion_val=0, refsig_channel=0)

Measure the maximum voluntary contraction (MVC).

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile with the reference signal.

TYPE: dict

how

showselect Ask the user to select the area where to calculate the MVC with a GUI.

all Calculate the MVC on the entire file.

TYPE: str {"showselect", "all"} DEFAULT: "showselect"

conversion_val

The conversion value to multiply the original reference signal. I.e., if the original reference signal is in kilogram (kg) and conversion_val=9.81, the output will be in Newton (N). If conversion_val=0 (default), the results will simply be in the original measure unit. conversion_val can be any custom int or float.

TYPE: float or int DEFAULT: 0

refsig_channel

The name of the reference signal channel (dataframe column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

RETURNS DESCRIPTION
mvc

The MVC value in the original (or converted) unit of measurement.

TYPE: float

See also
  • compute_rfd : calculate the RFD.
  • remove_offset : remove the offset from the REF_SIGNAL.
  • filter_refsig : low-pass filter REF_SIGNAL.

Examples:

Load the EMG file, remove reference signal offset and get MVC value.

>>> import openhdemg.library as emg
>>> emg_refsig = emg.askopenfile(filesource="OTB_REFSIG")
>>> offs_refsig = emg.remove_offset(emgfile=emg_refsig)
>>> mvc = emg.get_mvc(emgfile=offs_refsig )
>>> mvc
50.72

The process can be automated by bypassing the GUI and calculating the MVC of the entire file.

>>> import openhdemg.library as emg
>>> emg_refsig = emg.askopenfile(filesource="OTB_REFSIG")
>>> mvc = emg.get_mvc(emgfile=emg_refsig, how="all")
>>> print(mvc)
50.86


compute_rfd(emgfile, ms=[50, 100, 150, 200], startpoint=None, conversion_val=0, refsig_channel=0)

Calculate the RFD.

Rate of force development (RFD) is reported as X/Sec where "X" is the unit of measurement based on conversion_val.

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile with the reference signal.

TYPE: dict

ms

Milliseconds (ms). A list containing the ranges in ms to calculate the RFD.

TYPE: list DEFAULT: [50, 100, 150, 200]

startpoint

The starting point to calculate the RFD in samples, If None, the user will be requested to manually select the starting point.

TYPE: None or int DEFAULT: None

conversion_val

The conversion value to multiply the original reference signal. I.e., if the original reference signal is in kilogram (kg) and conversion_val=9.81, the output will be in Newton/Sec (N/Sec). If conversion_val=0 (default), the results will simply be Original measure unit/Sec. conversion_val can be any custom int or float.

TYPE: float or int DEFAULT: 0

refsig_channel

The name of the reference signal channel (dataframe column) to plot.

TYPE: int or str, Default 0 DEFAULT: 0

RETURNS DESCRIPTION
rfd

A pd.DataFrame containing the RFD at the different times.

TYPE: DataFrame

See also
  • get_mvif : measure the MViF.
  • remove_offset : remove the offset from the REF_SIGNAL.
  • filter_refsig : low-pass filter REF_SIGNAL.

Examples:

Load the EMG file, low-pass filter the reference signal and compute RFD.

>>> import openhdemg.library as emg
>>> emg_refsig = emg.askopenfile(filesource="OTB_REFSIG")
>>> filteredrefsig  = emg.filter_refsig(
...     emgfile=emg_refsig,
...     order=4,
...     cutoff=15,
... )
>>> rfd = emg.compute_rfd(
...     emgfile=filteredrefsig,
...     ms=[50, 100, 200],
...     conversion_val=9.81,
...     )
>>> rfd
        50         100        200
0  68.34342  79.296188  41.308215

The process can be automated by bypassing the GUI.

>>> import openhdemg.library as emg
>>> emg_refsig = emg.askopenfile(filesource="OTB_REFSIG")
>>> filteredrefsig  = emg.filter_refsig(
...     emgfile=emg_refsig,
...     order=4,
...     cutoff=15,
...     )
>>> rfd = emg.compute_rfd(
...     emgfile=filteredrefsig,
...     ms=[50, 100, 200],
...     startpoint=3568,
...     )
>>> rfd
        50         100        200
0  68.34342  79.296188  41.308215


compute_svr(emgfile, gammain=1 / 1.6, regparam=1 / 0.37, endpointweights_numpulses=5, endpointweights_magnitude=5, discontfiring_dur=1.0)

Fit MU discharge rates with Support Vector Regression, nonlinear regression.

Provides smooth and continous estimates of discharge rate useful for quantification and visualisation. Suggested hyperparameters and framework from Beauchamp et. al., 2022 https://doi.org/10.1088/1741-2552/ac4594

Author: James (Drew) Beauchamp

PARAMETER DESCRIPTION
emgfile

The dictionary containing the emgfile.

TYPE: dict

gammain

The kernel coefficient.

TYPE: float DEFAULT: 1/1.6

regparam

The regularization parameter, must be positive.

TYPE: float DEFAULT: 1/0.370

endpointweights_numpulses

Number of discharge instances at the start and end of MU firing to apply a weighting coefficient.

TYPE: int DEFAULT: 5

endpointweights_magnitude

The scaling factor applied to the number of pulses provided by endpointweights_numpulses. The scaling is applied to the regularization parameter, per sample. Larger values force the classifier to put more emphasis on the number of discharge instances at the start and end of firing provided by endpointweights_numpulses.

TYPE: int DEFAULT: 5

discontfiring_dur

Duration of time in seconds that defines an instnance of discontinuous firing. SVR fits will not be returned at points of discontinuity.

TYPE: int DEFAULT: 1

RETURNS DESCRIPTION
svrfits

A pd.DataFrame containing the smooth/continous MU discharge rates and corresponding time vectors.

TYPE: DataFrame

See also
  • compute_deltaf : quantify delta F via paired motor unit analysis.

Examples:

Quantify svr fits.

>>> import openhdemg.library as emg
>>> import pandas as pd
>>> emgfile = emg.emg_from_samplefile()
>>> emgfile = emg.sort_mus(emgfile=emgfile)
>>> svrfits = emg.compute_svr(emgfile)

Quick plot showing the results.

>>> smoothfits = pd.DataFrame(svrfits["gensvr"]).transpose()
>>> emg.plot_smoothed_dr(
>>>     emgfile,
>>>     smoothfits=smoothfits,
>>>     munumber="all",
>>>     addidr=False,
>>>     stack=True,
>>>     addrefsig=True,
>>> )